Detect storage_system_stack_trace_pipe_read_timeout_ms (compatibily f…

…or <23.5)

This storage_system_stack_trace_pipe_read_timeout_ms used for live
flamegraph feature, that shows live thread stack traces from
system.stack_trace. The setting itself useful to increase the
probability that the stacktrace will be captured and failed due to
timeout.

src/interpreter/clickhouse.rs

@@ -1,4 +1,7 @@
-use crate::interpreter::{options::ClickHouseOptions, ClickHouseAvailableQuirks, ClickHouseQuirks};
+use crate::interpreter::{
+ options::ClickHouseOptions, ClickHouseAvailableQuirks, ClickHouseCompatibility,
+ ClickHouseCompatibilitySettings, ClickHouseQuirks,
+};
 use anyhow::{Error, Result};
 use chrono::DateTime;
 use chrono_tz::Tz;
@@ -18,6 +21,7 @@ pub type Columns = Block<Complex>;
 pub struct ClickHouse {
  options: ClickHouseOptions,
  quirks: ClickHouseQuirks,
+ compatiblity: ClickHouseCompatibility,
 
  pool: Pool,
 }
@@ -135,10 +139,12 @@ impl ClickHouse {
  .fetch_all()
  .await?
  .get::<String, _>(0, 0)?;
- let quirks = ClickHouseQuirks::new(version);
+ let quirks = ClickHouseQuirks::new(version.clone());
+ let compatiblity = ClickHouseCompatibility::new(version.clone());
  return Ok(ClickHouse {
  options,
  quirks,
+ compatiblity,
  pool,
  });
  }
@@ -639,14 +645,17 @@ impl ClickHouse {
  FROM {}
  WHERE query_id IN ('{}')
  GROUP BY human_trace
- SETTINGS
- allow_introspection_functions=1,
- /* TODO: add settings support for clickhouse-rs, and use them with
- * is_important=false (for compatiblity with previous versions at least) */
- storage_system_stack_trace_pipe_read_timeout_ms=1000
+ SETTINGS allow_introspection_functions=1 {}
  "#,
  dbtable,
  query_ids.join("','"),
+ // TODO: add settings support for clickhouse-rs, and use them with
+ // is_important=false (for compatiblity with previous versions at least)
+ if self.compatiblity.has(ClickHouseCompatibilitySettings::storage_system_stack_trace_pipe_read_timeout_ms) {
+ ",storage_system_stack_trace_pipe_read_timeout_ms=1000"
+ } else {
+ ""
+ },
  ))
  .await;
  }

src/interpreter/clickhouse_compatibility.rs

@@ -0,0 +1,45 @@
+use semver::{Version, VersionReq};
+
+#[allow(non_camel_case_types)]
+#[derive(Debug, Clone, Copy)]
+pub enum ClickHouseCompatibilitySettings {
+ storage_system_stack_trace_pipe_read_timeout_ms = 1,
+}
+
+const SETTINGS: [(&str, ClickHouseCompatibilitySettings); 1] = [(
+ ">=23.5",
+ ClickHouseCompatibilitySettings::storage_system_stack_trace_pipe_read_timeout_ms,
+)];
+
+pub struct ClickHouseCompatibility {
+ mask: u64,
+}
+
+impl ClickHouseCompatibility {
+ pub fn new(version_string: String) -> Self {
+ // Version::parse() supports only x.y.z and nothing more.
+ let ver_maj_min_patch = version_string.split('.').collect::<Vec<&str>>()[0..3].join(".");
+ log::debug!("Version (maj.min.patch): {}", ver_maj_min_patch);
+
+ let version = Version::parse(ver_maj_min_patch.as_str())
+ .expect(&format!("Cannot parse version: {}", ver_maj_min_patch));
+ let mut mask: u64 = 0;
+
+ for setting in &SETTINGS {
+ let version_requirement = VersionReq::parse(setting.0).expect(&format!(
+ "Cannot parse version requirements for {:?}",
+ setting.1
+ ));
+ if version_requirement.matches(&version) {
+ mask |= setting.1 as u64;
+ log::warn!("Apply setting {:?}", setting.1);
+ }
+ }
+
+ return Self { mask };
+ }
+
+ pub fn has(&self, setting: ClickHouseCompatibilitySettings) -> bool {
+ return (self.mask & setting as u64) != 0;
+ }
+}

src/interpreter/mod.rs

@@ -1,6 +1,7 @@
 // pub for clickhouse::Columns
 mod background_runner;
 pub mod clickhouse;
+mod clickhouse_compatibility;
 mod clickhouse_quirks;
 mod context;
 mod query_process;
@@ -10,6 +11,8 @@ pub mod flamegraph;
 pub mod options;
 
 pub use clickhouse::ClickHouse;
+pub use clickhouse_compatibility::ClickHouseCompatibility;
+pub use clickhouse_compatibility::ClickHouseCompatibilitySettings;
 pub use clickhouse_quirks::ClickHouseAvailableQuirks;
 pub use clickhouse_quirks::ClickHouseQuirks;
 pub use context::Context;