clickhouse.rs

use crate::{
    common::RelativeDateTime,
    interpreter::{
        ClickHouseAvailableQuirks, ClickHouseQuirks,
        options::{ClickHouseOptions, LogsOrder},
    },
};
use anyhow::{Error, Result};
use chrono::{DateTime, Local};
use chrono_tz::Tz;
use clickhouse_rs::{
    Block, Options, Pool,
    types::{ColumnType, Complex, Enum8, Enum16, FromSql, Query, SqlType},
};
use futures_util::StreamExt;
use std::collections::HashMap;
use std::str::FromStr;

// TODO:
// - implement parsing using serde
// - replace clickhouse_rs::client_info::write() (with extend crate) to change the client name
// - escape parameters

pub type Columns = Block<Complex>;

// clickhouse-rs reads an Enum column as its bare integer (FromSql drops the name mapping) and a
// UUID as uuid::Uuid, so block.get::<String> fails on both; resolve the enum name from the Vec
// carried by the column type and format the UUID. String and LowCardinality(String) fall through
// to a plain String read (get coerces LC).
pub fn column_as_string<K: ColumnType>(block: &Block<K>, row: usize, name: &str) -> Result<String> {
    let column = block
        .columns()
        .iter()
        .find(|c| c.name() == name)
        .ok_or_else(|| Error::msg(format!("Cannot get {name} column")))?;
    fn name_of<T: PartialEq + ToString>(values: &[(String, T)], v: T) -> String {
        values
            .iter()
            .find(|(_, k)| *k == v)
            .map_or_else(|| v.to_string(), |(name, _)| name.clone())
    }
    Ok(match column.sql_type() {
        SqlType::Enum8(values) => name_of(&values, block.get::<Enum8, _>(row, name)?.internal()),
        SqlType::Enum16(values) => name_of(&values, block.get::<Enum16, _>(row, name)?.internal()),
        SqlType::Uuid => block.get::<uuid::Uuid, _>(row, name)?.to_string(),
        _ => block.get::<String, _>(row, name)?,
    })
}

pub struct ClickHouse {
    pub quirks: ClickHouseQuirks,
    // Server has use_shared_merge_tree_log_pipeline enabled (SharedMergeTree-backed system.*_log).
    // When true, system.*_log reads do not need clusterAllReplicas(): one replica sees all rows.
    shared_log_pipeline: bool,
    options: ClickHouseOptions,
    pool: Pool,
}

#[derive(Debug, PartialEq, Clone)]
#[allow(clippy::upper_case_acronyms)]
pub enum TraceType {
    CPU,
    Real,
    Memory,
    MemorySample,
    JemallocSample,
    ProfileEvent,
    MemoryAllocatedWithoutCheck,
}

#[derive(Debug, Clone)]
pub struct TextLogArguments {
    pub query_ids: Option<Vec<String>>,
    pub logger_names: Option<Vec<String>>,
    pub hostname: Option<String>,
    pub message_filter: Option<String>,
    pub max_level: Option<String>,
    pub start: DateTime<Local>,
    pub end: RelativeDateTime,
}

#[derive(Debug, Clone)]
pub struct PerfettoQueryScope {
    pub query_ids: Option<Vec<String>>,
    pub start: DateTime<Local>,
    pub end: DateTime<Local>,
}

#[derive(Default)]
pub struct ClickHouseServerCPU {
    pub count: u64,
    pub user: u64,
    pub system: u64,
}
/// NOTE: Likely misses threads for IO
#[derive(Default)]
pub struct ClickHouseServerThreadPools {
    pub merges_mutations: u64,
    pub fetches: u64,
    pub common: u64,
    pub moves: u64,
    pub schedule: u64,
    pub buffer_flush: u64,
    pub distributed: u64,
    pub message_broker: u64,
    pub backups: u64,
    pub io: u64,
    pub remote_io: u64,
    pub queries: u64,
}
#[derive(Default)]
pub struct ClickHouseServerThreads {
    pub os_total: u64,
    pub os_runnable: u64,
    pub tcp: u64,
    pub http: u64,
    pub interserver: u64,
    pub pools: ClickHouseServerThreadPools,
}
#[derive(Default)]
pub struct ClickHouseServerMemory {
    pub os_total: u64,
    pub resident: u64,

    pub tracked: u64,
    pub tables: u64,
    pub caches: u64,
    pub queries: u64,
    pub merges_mutations: u64,
    pub active_merges: u64,
    pub async_inserts: u64,
    pub dictionaries: u64,
    pub primary_keys: u64,
    pub fragmentation: u64,
    pub index_granularity: u64,
    pub io: u64,
}
/// May have duplicated accounting (due to bridges and stuff)
#[derive(Default)]
pub struct ClickHouseServerNetwork {
    pub send_bytes: u64,
    pub receive_bytes: u64,
}
#[derive(Default)]
pub struct ClickHouseServerUptime {
    pub _os: u64,
    pub server: u64,
}
/// May does not take into account some block devices (due to filter by sd*/nvme*/vd*)
#[derive(Default)]
pub struct ClickHouseServerBlockDevices {
    pub read_bytes: u64,
    pub write_bytes: u64,
}
#[derive(Default)]
pub struct ClickHouseServerStorages {
    pub buffer_bytes: u64,
    // Replace with bytes once [1] will be merged.
    //
    //   [1]: https://github.com/ClickHouse/ClickHouse/pull/50238
    pub distributed_insert_files: u64,
    pub total_rows: u64,
    pub total_bytes: u64,
}
#[derive(Default)]
pub struct ClickHouseServerRows {
    pub selected: u64,
    pub inserted: u64,
}
#[derive(Default)]
pub struct ClickHouseServerSummary {
    pub queries: u64,
    pub merges: u64,
    pub mutations: u64,
    pub replication_queue: u64,
    pub replication_queue_tries: u64,
    pub replication_max_absolute_delay: u64,
    pub fetches: u64,
    pub servers: u64,
    pub rows: ClickHouseServerRows,
    pub storages: ClickHouseServerStorages,
    pub uptime: ClickHouseServerUptime,
    pub memory: ClickHouseServerMemory,
    pub cpu: ClickHouseServerCPU,
    pub threads: ClickHouseServerThreads,
    pub network: ClickHouseServerNetwork,
    pub blkdev: ClickHouseServerBlockDevices,
    pub update_interval: u64,
}

pub struct QueryMetricRow {
    pub host_name: String,
    pub timestamp_ns: u64,
    pub memory_usage: i64,
    pub peak_memory_usage: i64,
    pub profile_events: HashMap<String, u64>,
}

pub struct MetricLogRow {
    pub timestamp_ns: u64,
    pub profile_events: HashMap<String, u64>,
    pub current_metrics: HashMap<String, i64>,
}

// Lookup by column index: by-name lookup is a linear scan in clickhouse-rs,
// which is quadratic over ~2K ProfileEvent_*/CurrentMetric_* columns.
fn metric_columns<'a, K: ColumnType>(block: &'a Block<K>, prefix: &str) -> Vec<(usize, &'a str)> {
    block
        .columns()
        .iter()
        .enumerate()
        .filter_map(|(idx, c)| c.name().strip_prefix(prefix).map(|name| (idx, name)))
        .collect()
}

fn block_timestamp_ns<K: ColumnType>(block: &Block<K>, i: usize, log: &str) -> Option<u64> {
    match block.get::<DateTime<Tz>, _>(i, "event_time_microseconds") {
        Ok(dt) => Some(dt.with_timezone(&Local).timestamp_nanos_opt().unwrap_or(0) as u64),
        Err(e) => {
            log::warn!("Perfetto: {} row {} event_time_microseconds: {}", log, i, e);
            None
        }
    }
}

pub fn parse_query_metric_log_block<K: ColumnType>(block: &Block<K>) -> Vec<QueryMetricRow> {
    let pe_columns = metric_columns(block, "ProfileEvent_");

    let mut rows = Vec::with_capacity(block.row_count());
    for i in 0..block.row_count() {
        let mut profile_events = HashMap::new();
        for &(idx, name) in &pe_columns {
            let value: u64 = block.get(i, idx).unwrap_or(0);
            if value != 0 {
                profile_events.insert(name.to_string(), value);
            }
        }
        let Some(ts_ns) = block_timestamp_ns(block, i, "query_metric_log") else {
            continue;
        };
        rows.push(QueryMetricRow {
            host_name: block.get(i, "host_name").unwrap_or_default(),
            timestamp_ns: ts_ns,
            memory_usage: block.get(i, "memory_usage").unwrap_or(0),
            peak_memory_usage: block.get(i, "peak_memory_usage").unwrap_or(0),
            profile_events,
        });
    }
    rows
}

pub fn parse_metric_log_block<K: ColumnType>(block: &Block<K>) -> Vec<MetricLogRow> {
    let pe_columns = metric_columns(block, "ProfileEvent_");
    let cm_columns = metric_columns(block, "CurrentMetric_");

    let mut rows = Vec::with_capacity(block.row_count());
    for i in 0..block.row_count() {
        let Some(ts_ns) = block_timestamp_ns(block, i, "metric_log") else {
            continue;
        };
        let mut profile_events = HashMap::new();
        for &(idx, name) in &pe_columns {
            let value: u64 = block.get(i, idx).unwrap_or(0);
            if value != 0 {
                profile_events.insert(name.to_string(), value);
            }
        }
        let mut current_metrics = HashMap::new();
        for &(idx, name) in &cm_columns {
            let value: i64 = block.get(i, idx).unwrap_or(0);
            if value != 0 {
                current_metrics.insert(name.to_string(), value);
            }
        }
        rows.push(MetricLogRow {
            timestamp_ns: ts_ns,
            profile_events,
            current_metrics,
        });
    }
    rows
}

fn collect_values<'b, T: FromSql<'b>>(block: &'b Columns, column: &str) -> Vec<T> {
    return (0..block.row_count())
        .map(|i| block.get(i, column).unwrap())
        .collect();
}

const CHDIG_CLIENT_NAME: [&str; 2] = ["chdig", env!("CARGO_PKG_VERSION")];
fn get_client_name() -> String {
    return CHDIG_CLIENT_NAME.join("-");
}

impl ClickHouse {
    pub async fn new(options: ClickHouseOptions) -> Result<Self> {
        let url = format!(
            "{}&client_name={}",
            options.url.clone().unwrap(),
            get_client_name()
        );
        let connect_options: Options = Options::from_str(&url)?.with_setting(
            "storage_system_stack_trace_pipe_read_timeout_ms",
            1000,
            /* is_important= */ false,
        );
        let pool = Pool::new(connect_options);

        let mut handle = pool.get_handle().await.map_err(|e| {
            Error::msg(format!(
                "Cannot connect to ClickHouse at {} ({})",
                options.url_safe, e
            ))
        })?;

        let version = if let Some(override_version) = &options.server_version {
            override_version.clone()
        } else {
            let version = handle
                .query("SELECT version()")
                .fetch_all()
                .await?
                .get::<String, _>(0, 0)?;

            // Get VERSION_DESCRIBE from system.build_options for full version info (only build_options
            // include version prefix, i.e. -stable/-testing)
            handle
                .query("SELECT value FROM system.build_options WHERE name = 'VERSION_DESCRIBE'")
                .fetch_all()
                .await?
                .get::<String, _>(0, 0)
                .unwrap_or_else(|_| version.clone())
        };

        let quirks = ClickHouseQuirks::new(version);

        // SMT-backed system.*_log (ClickHouse Cloud) exposes all replicas' rows through any single
        // replica, so clusterAllReplicas() is pure overhead there. The setting is off by default
        // and on self-hosted clusters, so we silently fall back to the cluster-wrapped path.
        let shared_log_pipeline = handle
            .query(
                "SELECT value FROM system.server_settings \
                 WHERE name = 'use_shared_merge_tree_log_pipeline'",
            )
            .fetch_all()
            .await
            .ok()
            .filter(|block| block.row_count() > 0)
            .and_then(|block| block.get::<String, _>(0, 0).ok())
            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
            .unwrap_or(false);
        if shared_log_pipeline {
            log::info!(
                "SharedMergeTree log pipeline detected, skipping clusterAllReplicas() for system.*_log"
            );
        }

        return Ok(ClickHouse {
            quirks,
            shared_log_pipeline,
            options,
            pool,
        });
    }

    pub fn version(&self) -> String {
        return self.quirks.get_version();
    }

    pub async fn get_slow_query_log(
        &self,
        filter: &String,
        start: RelativeDateTime,
        end: RelativeDateTime,
        limit: u64,
        selected_host: Option<&String>,
    ) -> Result<Columns> {
        let dbtable = self.get_log_table_name("query_log");
        let host_filter = self.get_log_host_filter_clause(selected_host);
        return self
            .execute(
                format!(
                    r#"
                    WITH
                        {start} AS start_,
                        {end}   AS end_,
                        slow_queries_ids AS (
                            SELECT DISTINCT initial_query_id
                            FROM {db_table}
                            WHERE
                                event_date BETWEEN toDate(start_) AND toDate(end_) AND
                                event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                                is_initial_query AND
                                /* To make query faster */
                                query_duration_ms > 1e3
                                {filter}
                                {internal}
                                {host_filter}
                            ORDER BY query_duration_ms DESC
                            LIMIT {limit}
                        )
                    SELECT
                        mapKeys(ProfileEvents)   AS `ProfileEvents.Names`,
                        mapValues(ProfileEvents) AS `ProfileEvents.Values`,
                        mapKeys(Settings)        AS `Settings.Names`,
                        mapValues(Settings)      AS `Settings.Values`,
                        {peak_threads_usage} AS peak_threads_usage,
                        // Compatibility with system.processlist
                        memory_usage::Int64 AS peak_memory_usage,
                        query_duration_ms/1e3 AS elapsed,
                        user,
                        initial_user,
                        exception,
                        is_initial_query,
                        (exception_code = 394)::UInt8 AS is_cancelled,
                        initial_query_id,
                        query_id,
                        hostname as host_name,
                        current_database,
                        query_start_time_microseconds,
                        event_time_microseconds AS query_end_time_microseconds,
                        toValidUTF8(query) AS original_query,
                        normalizeQuery(query) AS normalized_query,
                        normalized_query_hash
                    FROM {db_table}
                    PREWHERE
                        event_date BETWEEN toDate(start_) AND toDate(end_) AND
                        event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                        type != 'QueryStart' AND
                        initial_query_id GLOBAL IN slow_queries_ids
                "#,
                    start = start.to_sql_datetime_64().ok_or(Error::msg("Invalid start"))?,
                    end = end.to_sql_datetime_64().ok_or(Error::msg("Invalid end"))?,
                    db_table = dbtable,
                    peak_threads_usage = if self.quirks.has(ClickHouseAvailableQuirks::QueryLogPeakThreadsUsage) {
                        "peak_threads_usage"
                    } else {
                        "length(thread_ids)"
                    },
                    internal = self.get_internal_filter_clause(),
                    filter = if !filter.is_empty() {
                        format!("AND (client_hostname LIKE '{0}' OR log_comment LIKE '{0}' OR os_user LIKE '{0}' OR user LIKE '{0}' OR initial_user LIKE '{0}' OR client_name LIKE '{0}' OR query_id LIKE '{0}' OR query LIKE '{0}' OR current_database LIKE '{0}' OR toString(normalized_query_hash) LIKE '{0}')", &filter)
                    } else {
                        "".to_string()
                    },
                    host_filter = host_filter,
                )
                .as_str(),
            )
            .await;
    }

    pub async fn get_last_query_log(
        &self,
        filter: &String,
        start: RelativeDateTime,
        end: RelativeDateTime,
        limit: u64,
        selected_host: Option<&String>,
    ) -> Result<Columns> {
        // TODO:
        // - propagate sort order from the table
        // - distributed_group_by_no_merge=2 is broken for this query with WINDOW function
        let dbtable = self.get_log_table_name("query_log");
        let host_filter = self.get_log_host_filter_clause(selected_host);
        return self
            .execute(
                format!(
                    r#"
                    WITH
                        {start} AS start_,
                        {end}   AS end_,
                        last_queries_ids AS (
                            SELECT DISTINCT initial_query_id
                            FROM {db_table}
                            WHERE
                                event_date BETWEEN toDate(start_) AND toDate(end_) AND
                                event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                                is_initial_query
                                {filter}
                                {internal}
                                {host_filter}
                            ORDER BY event_date DESC, event_time DESC
                            LIMIT {limit}
                        )
                    SELECT
                        mapKeys(ProfileEvents)   AS `ProfileEvents.Names`,
                        mapValues(ProfileEvents) AS `ProfileEvents.Values`,
                        mapKeys(Settings)        AS `Settings.Names`,
                        mapValues(Settings)      AS `Settings.Values`,
                        {peak_threads_usage} AS peak_threads_usage,
                        // Compatibility with system.processlist
                        memory_usage::Int64 AS peak_memory_usage,
                        query_duration_ms/1e3 AS elapsed,
                        user,
                        initial_user,
                        exception,
                        is_initial_query,
                        (exception_code = 394)::UInt8 AS is_cancelled,
                        initial_query_id,
                        query_id,
                        hostname as host_name,
                        current_database,
                        query_start_time_microseconds,
                        event_time_microseconds AS query_end_time_microseconds,
                        toValidUTF8(query) AS original_query,
                        normalizeQuery(query) AS normalized_query,
                        normalized_query_hash
                    FROM {db_table}
                    PREWHERE
                        event_date BETWEEN toDate(start_) AND toDate(end_) AND
                        event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                        type != 'QueryStart' AND
                        initial_query_id GLOBAL IN last_queries_ids
                "#,
                    start = start.to_sql_datetime_64().ok_or(Error::msg("Invalid start"))?,
                    end = end.to_sql_datetime_64().ok_or(Error::msg("Invalid end"))?,
                    db_table = dbtable,
                    peak_threads_usage = if self.quirks.has(ClickHouseAvailableQuirks::QueryLogPeakThreadsUsage) {
                        "peak_threads_usage"
                    } else {
                        "length(thread_ids)"
                    },
                    internal = self.get_internal_filter_clause(),
                    filter = if !filter.is_empty() {
                        format!("AND (client_hostname LIKE '{0}' OR log_comment LIKE '{0}' OR os_user LIKE '{0}' OR user LIKE '{0}' OR initial_user LIKE '{0}' OR client_name LIKE '{0}' OR query_id LIKE '{0}' OR query LIKE '{0}' OR current_database LIKE '{0}' OR toString(normalized_query_hash) LIKE '{0}')", &filter)
                    } else {
                        "".to_string()
                    },
                    host_filter = host_filter,
                )
                .as_str(),
            )
            .await;
    }

    pub async fn get_processlist(
        &self,
        filter: String,
        limit: u64,
        selected_host: Option<&String>,
    ) -> Result<Columns> {
        let dbtable = self.get_live_table_name("processes");
        let host_filter = self.get_host_filter_clause(selected_host);
        return self
            .execute(
                format!(
                    r#"
                    SELECT
                        mapKeys(ProfileEvents)   AS `ProfileEvents.Names`,
                        mapValues(ProfileEvents) AS `ProfileEvents.Values`,
                        mapKeys(Settings)        AS `Settings.Names`,
                        mapValues(Settings)      AS `Settings.Values`,
                        {peak_threads_usage} AS peak_threads_usage,
                        peak_memory_usage,
                        elapsed / {q} AS elapsed,
                        user,
                        initial_user,
                        '' AS exception,
                        is_initial_query,
                        is_cancelled,
                        initial_query_id,
                        query_id,
                        hostName() AS host_name,
                        {current_database} AS current_database,
                        /* NOTE: now64()/elapsed does not have enough precision to handle starting
                         * time properly, while this column is used for querying system.text_log,
                         * and it should be the smallest time that we are looking for */
                        (now64(6) - elapsed - 1) AS query_start_time_microseconds,
                        now64(6) AS query_end_time_microseconds,
                        toValidUTF8(query) AS original_query,
                        normalizeQuery(query) AS normalized_query,
                        normalizedQueryHash(query) AS normalized_query_hash
                    FROM {}
                    WHERE 1
                    {filter}
                    {internal}
                    {host_filter}
                    LIMIT {limit}
                "#,
                    dbtable,
                    q = if self.quirks.has(ClickHouseAvailableQuirks::ProcessesElapsed) {
                        10
                    } else {
                        1
                    },
                    current_database = if self.quirks.has(ClickHouseAvailableQuirks::ProcessesCurrentDatabase) {
                        // This is required for EXPLAIN (available since 20.6),
                        // so EXPLAIN with non-default current_database will be broken from processes view.
                        "'default'"
                    } else {
                        "current_database"
                    },
                    internal = self.get_internal_filter_clause(),
                    filter = if !filter.is_empty() {
                        format!("AND (client_hostname LIKE '{0}' OR Settings['log_comment'] LIKE '{0}' OR os_user LIKE '{0}' OR user LIKE '{0}' OR initial_user LIKE '{0}' OR client_name LIKE '{0}' OR query_id LIKE '{0}' OR query LIKE '{0}' OR current_database LIKE '{0}' OR toString(normalizedQueryHash(query)) LIKE '{0}')", &filter)
                    } else {
                        "".to_string()
                    },
                    peak_threads_usage = if self.quirks.has(ClickHouseAvailableQuirks::ProcessesPeakThreadsUsage) {
                        "peak_threads_usage"
                    } else {
                        "length(thread_ids)"
                    },
                    host_filter = host_filter,
                )
                .as_str(),
            )
            .await;
    }

    pub async fn get_summary(
        &self,
        selected_host: Option<&String>,
    ) -> Result<ClickHouseServerSummary> {
        let host_filter = self.get_host_filter_clause(selected_host);
        let host_where = if host_filter.is_empty() {
            String::new()
        } else {
            format!(" WHERE {}", &host_filter[4..]) // Remove leading "AND "
        };

        let memory_index_granularity_trait = if self.quirks.has(ClickHouseAvailableQuirks::AsynchronousMetricsTotalIndexGranularityBytesInMemoryAllocated) {
            format!("(SELECT sum(index_granularity_bytes_in_memory_allocated) FROM {}{}) AS memory_index_granularity_", self.get_live_table_name("parts"), host_where)
        } else {
            "0::UInt64 AS memory_index_granularity_".to_string()
        };

        // NOTE: metrics (but not all of them) are deltas, so chdig do not need to reimplement this logic by itself.
        let block = self
            .execute(
                &format!(
                    r#"
                    WITH
                        -- memory detalization
                        (SELECT sum(CAST(value AS UInt64)) FROM {metrics} WHERE metric = 'MemoryTracking' {host_filter_and}) AS memory_tracked_,
                        (SELECT sum(CAST(value AS UInt64)) FROM {metrics} WHERE metric = 'MergesMutationsMemoryTracking' {host_filter_and}) AS memory_merges_mutations_,
                        (SELECT sum(total_bytes) FROM {tables} WHERE engine IN ('Join','Memory','Buffer','Set') {host_filter_and}) AS memory_tables_,
                        (SELECT sum(CAST(value AS UInt64)) FROM {asynchronous_metrics} WHERE metric LIKE '%CacheBytes' AND metric NOT LIKE '%Filesystem%' {host_filter_and}) AS memory_async_metrics_caches_,
                        (SELECT sum(CAST(value AS UInt64)) FROM {metrics} WHERE
                            metric NOT LIKE '%Filesystem%' AND
                            (metric LIKE '%CacheBytes' OR metric IN ('IcebergMetadataFilesCacheSize', 'VectorSimilarityIndexCacheSize'))
                            {host_filter_and}
                        ) AS memory_metrics_caches_,
                        (SELECT sum(CAST(memory_usage AS UInt64)) FROM {processes} {host_filter_where})                              AS memory_queries_,
                        (SELECT sum(CAST(memory_usage AS UInt64)) FROM {merges} {host_filter_where})                                 AS memory_active_merges_,
                        (SELECT sum(bytes_allocated) FROM {dictionaries} {host_filter_where})                                        AS memory_dictionaries_,
                        (SELECT sum(total_bytes) FROM {async_inserts} {host_filter_where})                                           AS memory_async_inserts_,
                        {memory_index_granularity_trait},
                        (SELECT count() FROM {one} {host_filter_where})                                                              AS servers_,
                        (SELECT count() FROM {replication_queue} {host_filter_where})                                                AS replication_queue_,
                        (SELECT sum(num_tries) FROM {replication_queue} {host_filter_where})                                         AS replication_queue_tries_,
                        (SELECT [sum(total_rows), sum(total_bytes)] FROM (
                            SELECT
                                if(engine LIKE 'Shared%', max(total_rows), sum(total_rows)) AS total_rows,
                                if(engine LIKE 'Shared%', max(total_bytes), sum(total_bytes)) AS total_bytes
                            FROM {tables}
                            WHERE has_own_data = 1 {host_filter_and}
                            GROUP BY database, name, engine
                        )) AS storage_totals_
                    SELECT
                        assumeNotNull(memory_tracked_)                           AS memory_tracked,
                        assumeNotNull(memory_merges_mutations_)                  AS memory_merges_mutations,
                        assumeNotNull(memory_tables_)                            AS memory_tables,
                        assumeNotNull(memory_async_metrics_caches_) + assumeNotNull(memory_metrics_caches_) AS memory_caches,
                        assumeNotNull(memory_queries_)                           AS memory_queries,
                        assumeNotNull(memory_active_merges_)                     AS memory_active_merges,
                        assumeNotNull(memory_dictionaries_)                      AS memory_dictionaries,
                        assumeNotNull(memory_async_inserts_)                     AS memory_async_inserts,
                        assumeNotNull(servers_)                                  AS servers,
                        assumeNotNull(replication_queue_)                        AS replication_queue,
                        assumeNotNull(replication_queue_tries_)                  AS replication_queue_tries,
                        assumeNotNull(storage_totals_[1])::UInt64               AS storage_total_rows,
                        assumeNotNull(storage_totals_[2])::UInt64              AS storage_total_bytes,

                        max2(assumeNotNull(memory_index_granularity_), asynchronous_metrics.memory_index_granularity)::UInt64 AS memory_index_granularity,

                        asynchronous_metrics.*,
                        events.*,
                        metrics.*
                    FROM
                    (
                        WITH
                            -- exclude MD/LVM
                            metric LIKE '%_sd%' OR metric LIKE '%_nvme%' OR metric LIKE '%_vd%' AS is_disk,
                            metric LIKE '%vlan%' AS is_vlan
                        -- NOTE: cast should be after aggregation function since the type is Float64
                        SELECT
                            CAST(minIf(value, metric == 'OSUptime') AS UInt64)       AS os_uptime,
                            CAST(min(uptime()) AS UInt64)                            AS uptime,
                            -- memory
                            CAST(coalesce(sumIfOrNull(value, metric == 'CGroupMemoryTotal' and value > 0), sumIf(value, metric == 'OSMemoryTotal')) AS UInt64) AS os_memory_total,
                            CAST(sumIf(value, metric == 'MemoryResident') AS UInt64) AS memory_resident,
                            -- May differs from primary_key_bytes_in_memory_allocated from
                            -- system.parts, since it takes into account only active parts
                            CAST(sumIf(value,
                                metric == 'TotalPrimaryKeyBytesInMemoryAllocated'
                                OR metric == 'TotalProjectionPrimaryKeyBytesInMemoryAllocated'
                            ) AS UInt64) AS memory_primary_keys,
                            CAST(sumIf(value,
                                metric == 'TotalIndexGranularityBytesInMemoryAllocated'
                                OR metric == 'TotalProjectionIndexGranularityBytesInMemoryAllocated'
                            ) AS UInt64) AS memory_index_granularity,
                            CAST((
                                sumIf(value, metric == 'jemalloc.resident') -
                                sumIf(value, metric == 'jemalloc.allocated')
                            ) AS UInt64) AS memory_fragmentation,
                            -- cpu
                            CAST(
                                max2(
                                    countIf(metric LIKE 'CPUFrequencyMHz%'),
                                    sumIf(value, metric = 'CGroupMaxCPU')
                                )
                            AS UInt64) AS cpu_count,
                            CAST(
                                max2(
                                    sumIf(value, metric LIKE 'OSUserTimeCPU%'),
                                    sumIf(value, metric = 'OSUserTime')
                                )
                            AS UInt64) AS cpu_user,
                            CAST(
                                max2(
                                    sumIf(value, metric LIKE 'OSSystemTimeCPU%'),
                                    sumIf(value, metric = 'OSSystemTime')
                                )
                            AS UInt64) AS cpu_system,
                            -- threads detalization
                            CAST(sumIf(value, metric = 'HTTPThreads') AS UInt64)             AS threads_http,
                            CAST(sumIf(value, metric = 'TCPThreads') AS UInt64)              AS threads_tcp,
                            CAST(sumIf(value, metric = 'OSThreadsTotal') AS UInt64)          AS threads_os_total,
                            CAST(sumIf(value, metric = 'OSThreadsRunnable') AS UInt64)       AS threads_os_runnable,
                            CAST(sumIf(value, metric = 'InterserverThreads') AS UInt64)      AS threads_interserver,
                            -- network
                            CAST(sumIf(value, metric LIKE 'NetworkSendBytes%' AND NOT is_vlan) AS UInt64)    AS net_send_bytes,
                            CAST(sumIf(value, metric LIKE 'NetworkReceiveBytes%' AND NOT is_vlan) AS UInt64) AS net_receive_bytes,
                            -- block devices
                            CAST(sumIf(value, metric LIKE 'BlockReadBytes%' AND is_disk) AS UInt64)      AS block_read_bytes,
                            CAST(sumIf(value, metric LIKE 'BlockWriteBytes%' AND is_disk) AS UInt64)     AS block_write_bytes,
                            -- replication
                            CAST(maxIf(value, metric == 'ReplicasMaxAbsoluteDelay') AS UInt64) AS replication_max_absolute_delay,
                            -- update intervals
                            CAST(anyLastIf(value, metric == 'AsynchronousMetricsUpdateInterval') AS UInt64) AS metrics_update_interval
                        FROM {asynchronous_metrics}
                        {host_filter_where}
                    ) as asynchronous_metrics,
                    (
                        SELECT
                            sumIf(CAST(value AS UInt64), event == 'IOBufferAllocBytes') AS memory_io,
                            sumIf(CAST(value AS UInt64), event == 'SelectedRows') AS selected_rows,
                            sumIf(CAST(value AS UInt64), event == 'InsertedRows') AS inserted_rows
                        FROM {events}
                        {host_filter_where}
                    ) as events,
                    (
                        SELECT
                            sumIf(CAST(value AS UInt64), metric == 'Query') AS queries,
                            sumIf(CAST(value AS UInt64), metric == 'Merge') AS merges,
                            sumIf(CAST(value AS UInt64), metric == 'PartMutation') AS mutations,
                            sumIf(CAST(value AS UInt64), metric == 'ReplicatedFetch') AS fetches,

                            sumIf(CAST(value AS UInt64), metric == 'StorageBufferBytes') AS storage_buffer_bytes,
                            sumIf(CAST(value AS UInt64), metric == 'DistributedFilesToInsert') AS storage_distributed_insert_files,

                            sumIf(CAST(value AS UInt64), metric == 'BackgroundMergesAndMutationsPoolTask')    AS threads_merges_mutations,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundFetchesPoolTask')               AS threads_fetches,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundCommonPoolTask')                AS threads_common,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundMovePoolTask')                  AS threads_moves,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundSchedulePoolTask')              AS threads_schedule,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundBufferFlushSchedulePoolTask')   AS threads_buffer_flush,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundDistributedSchedulePoolTask')   AS threads_distributed,
                            sumIf(CAST(value AS UInt64), metric == 'BackgroundMessageBrokerSchedulePoolTask') AS threads_message_broker,
                            sumIf(CAST(value AS UInt64), metric IN (
                                'BackupThreadsActive',
                                'RestoreThreadsActive',
                                'BackupsIOThreadsActive'
                            )) AS threads_backups,
                            sumIf(CAST(value AS UInt64), metric IN (
                                'DiskObjectStorageAsyncThreadsActive',
                                'ThreadPoolRemoteFSReaderThreadsActive',
                                'StorageS3ThreadsActive'
                            )) AS threads_remote_io,
                            sumIf(CAST(value AS UInt64), metric IN (
                                'IOThreadsActive',
                                'IOWriterThreadsActive',
                                'IOPrefetchThreadsActive',
                                'MarksLoaderThreadsActive'
                            )) AS threads_io,
                            sumIf(CAST(value AS UInt64), metric IN (
                                'QueryPipelineExecutorThreadsActive',
                                'QueryThread',
                                'AggregatorThreadsActive',
                                'StorageDistributedThreadsActive',
                                'DestroyAggregatesThreadsActive'
                            )) AS threads_queries
                        FROM {metrics}
                        {host_filter_where}
                    ) as metrics
                    SETTINGS enable_global_with_statement=0
                "#,
                    metrics=self.get_live_table_name("metrics"),
                    events=self.get_live_table_name("events"),
                    tables=self.get_live_table_name("tables"),
                    processes=self.get_live_table_name("processes"),
                    merges=self.get_live_table_name("merges"),
                    async_inserts=self.get_live_table_name("asynchronous_inserts"),
                    replication_queue=self.get_live_table_name("replication_queue"),
                    dictionaries=self.get_live_table_name("dictionaries"),
                    asynchronous_metrics=self.get_live_table_name("asynchronous_metrics"),
                    one=self.get_live_table_name("one"),

                    memory_index_granularity_trait=memory_index_granularity_trait,
                    host_filter_where=host_where,
                    host_filter_and=host_filter,
                )
            )
            .await?;

        let get = |key: &str| {
            // By subquery.column
            if let Ok(value) = block.get::<u64, _>(0, key) {
                return value;
            }

            let parts = key.split(".").collect::<Vec<&str>>();
            assert!(parts.len() <= 2);
            // By column
            return block.get::<u64, _>(0, parts[parts.len() - 1]).expect(key);
        };

        return Ok(ClickHouseServerSummary {
            queries: get("metrics.queries"),
            merges: get("metrics.merges"),
            mutations: get("metrics.mutations"),
            replication_queue: get("replication_queue"),
            replication_queue_tries: get("replication_queue_tries"),
            replication_max_absolute_delay: get(
                "asynchronous_metrics.replication_max_absolute_delay",
            ),
            fetches: get("metrics.fetches"),
            servers: get("servers"),

            uptime: ClickHouseServerUptime {
                _os: get("asynchronous_metrics.os_uptime"),
                server: get("asynchronous_metrics.uptime"),
            },

            rows: ClickHouseServerRows {
                selected: get("events.selected_rows"),
                inserted: get("events.inserted_rows"),
            },

            storages: ClickHouseServerStorages {
                buffer_bytes: get("metrics.storage_buffer_bytes"),
                distributed_insert_files: get("metrics.storage_distributed_insert_files"),
                total_rows: get("storage_total_rows"),
                total_bytes: get("storage_total_bytes"),
            },

            memory: ClickHouseServerMemory {
                os_total: get("asynchronous_metrics.os_memory_total"),
                resident: get("asynchronous_metrics.memory_resident"),

                tracked: get("memory_tracked"),
                merges_mutations: get("memory_merges_mutations"),
                tables: get("memory_tables"),
                caches: get("memory_caches"),
                queries: get("memory_queries"),
                active_merges: get("memory_active_merges"),
                async_inserts: get("memory_async_inserts"),
                dictionaries: get("memory_dictionaries"),
                primary_keys: get("asynchronous_metrics.memory_primary_keys"),
                fragmentation: get("asynchronous_metrics.memory_fragmentation"),
                index_granularity: get("memory_index_granularity"),
                io: get("events.memory_io"),
            },

            cpu: ClickHouseServerCPU {
                count: get("asynchronous_metrics.cpu_count"),
                user: get("asynchronous_metrics.cpu_user"),
                system: get("asynchronous_metrics.cpu_system"),
            },

            threads: ClickHouseServerThreads {
                os_total: get("asynchronous_metrics.threads_os_total"),
                os_runnable: get("asynchronous_metrics.threads_os_runnable"),
                http: get("asynchronous_metrics.threads_http"),
                tcp: get("asynchronous_metrics.threads_tcp"),
                interserver: get("asynchronous_metrics.threads_interserver"),
                pools: ClickHouseServerThreadPools {
                    merges_mutations: get("metrics.threads_merges_mutations"),
                    fetches: get("metrics.threads_fetches"),
                    common: get("metrics.threads_common"),
                    moves: get("metrics.threads_moves"),
                    schedule: get("metrics.threads_schedule"),
                    buffer_flush: get("metrics.threads_buffer_flush"),
                    distributed: get("metrics.threads_distributed"),
                    message_broker: get("metrics.threads_message_broker"),
                    backups: get("metrics.threads_backups"),
                    io: get("metrics.threads_io"),
                    remote_io: get("metrics.threads_remote_io"),
                    queries: get("metrics.threads_queries"),
                },
            },

            network: ClickHouseServerNetwork {
                send_bytes: get("asynchronous_metrics.net_send_bytes"),
                receive_bytes: get("asynchronous_metrics.net_receive_bytes"),
            },

            blkdev: ClickHouseServerBlockDevices {
                read_bytes: get("asynchronous_metrics.block_read_bytes"),
                write_bytes: get("asynchronous_metrics.block_write_bytes"),
            },

            update_interval: get("asynchronous_metrics.metrics_update_interval"),
        });
    }

    pub async fn kill_query(&self, query_id: &str) -> Result<()> {
        let query = if let Some(cluster) = &self.options.cluster {
            format!(
                "KILL QUERY ON CLUSTER {} WHERE query_id = '{}' SYNC",
                cluster, query_id
            )
        } else {
            format!("KILL QUERY WHERE query_id = '{}' SYNC", query_id)
        };
        return self.execute_simple(&query).await;
    }

    pub async fn execute_query(&self, database: &str, query: &str) -> Result<()> {
        self.execute_simple(&format!("USE {}", database)).await?;
        return self.execute_simple(query).await;
    }

    pub async fn explain_syntax(
        &self,
        database: &str,
        query: &str,
        settings: &HashMap<String, String>,
    ) -> Result<Vec<String>> {
        return self
            .explain("SYNTAX", database, query, Some(settings))
            .await;
    }

    pub async fn explain_plan(&self, database: &str, query: &str) -> Result<Vec<String>> {
        return self.explain("PLAN actions=1", database, query, None).await;
    }

    pub async fn explain_pipeline(&self, database: &str, query: &str) -> Result<Vec<String>> {
        return self.explain("PIPELINE", database, query, None).await;
    }

    pub async fn explain_pipeline_graph(&self, database: &str, query: &str) -> Result<Vec<String>> {
        return self
            .explain("PIPELINE graph=1", database, query, None)
            .await;
    }

    // NOTE: can we benefit from json=1?
    pub async fn explain_plan_indexes(&self, database: &str, query: &str) -> Result<Vec<String>> {
        return self.explain("PLAN indexes=1", database, query, None).await;
    }

    pub async fn show_create_table(&self, database: &str, table: &str) -> Result<String> {
        let result = self
            .execute(&format!("SHOW CREATE TABLE {}.{}", database, table))
            .await?;
        let statement: String = collect_values(&result, "statement")
            .into_iter()
            .next()
            .unwrap_or_default();
        return Ok(statement);
    }

    async fn explain(
        &self,
        what: &str,
        database: &str,
        query: &str,
        settings: Option<&HashMap<String, String>>,
    ) -> Result<Vec<String>> {
        self.execute_simple(&format!("USE {}", database)).await?;

        // The query's own settings are passed through the protocol rather than
        // appended as a SETTINGS clause: the query may already carry its own
        // SETTINGS, and a second one would be dropped from EXPLAIN SYNTAX output.
        let mut explain = Query::new(format!("EXPLAIN {} {}", what, query));
        if let Some(settings) = settings {
            for (name, value) in settings {
                explain = explain.with_setting(name, value.as_str(), false);
            }
        }

        return Ok(collect_values(&self.execute(explain).await?, "explain"));
    }

    /// Stream the text_log rows block-by-block into `on_block` (see
    /// execute_for_each() for the semantics).
    pub async fn get_query_logs(
        &self,
        args: &TextLogArguments,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        // TODO:
        // - optional flush, but right now it gives "blocks should not be empty." error
        //   self.execute("SYSTEM FLUSH LOGS").await;
        // - configure time interval
        //
        // NOTE:
        // - we cannot use LIVE VIEW, since
        //   a) they are pretty complex
        //   b) it does not work in case we monitor the whole cluster

        let dbtable = self.get_log_table_name("text_log");
        let order = if self.options.logs_order == LogsOrder::Desc {
            "DESC"
        } else {
            "ASC"
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({}) AS start_time_,
                        {} AS end_time_
                    SELECT
                        hostname AS host_name,
                        event_time,
                        event_time_microseconds,
                        thread_id,
                        level,
                        logger_name,
                        query_id,
                        message
                    FROM {}
                    WHERE
                            event_date >= toDate(start_time_) AND event_time >= toDateTime(start_time_) AND event_time_microseconds > start_time_
                        AND event_date <= toDate(end_time_)   AND event_time <= toDateTime(end_time_)   AND event_time_microseconds <= end_time_
                        {}
                        {}
                        {}
                        {}
                        {}
                    ORDER BY event_date {order}, event_time {order}, event_time_microseconds {order}
                    LIMIT {}
                    "#,
            args.start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start time"))?,
            args.end
                .to_sql_datetime_64()
                .ok_or(Error::msg("Invalid end time"))?,
            dbtable,
            if let Some(query_ids) = &args.query_ids {
                format!("AND query_id IN ('{}')", query_ids.join("','"))
            } else {
                "".into()
            },
            if let Some(logger_names) = &args.logger_names {
                format!(
                    "AND ({})",
                    logger_names
                        .iter()
                        .map(|l| format!("logger_name LIKE '{}'", l))
                        .collect::<Vec<_>>()
                        .join(" OR ")
                )
            } else {
                "".into()
            },
            if let Some(hostname) = &args.hostname {
                format!(
                    "AND (hostName() = '{0}' OR hostname = '{0}')",
                    hostname.replace('\'', "''")
                )
            } else {
                "".into()
            },
            if let Some(message_filter) = &args.message_filter {
                format!("AND message LIKE '%{}%'", message_filter)
            } else {
                "".into()
            },
            if let Some(max_level) = &args.max_level {
                format!("AND level <= '{}'", max_level)
            } else {
                "".into()
            },
            self.options.limit,
        );
        return self.execute_for_each_raw(&sql, on_block).await;
    }

    /// Return query flamegraph in pyspy format for flameshow.
    /// It is the same format as TSV, but with ' ' delimiter between symbols and weight.
    pub async fn get_flamegraph(
        &self,
        trace_type: TraceType,
        query_ids: Option<&[String]>,
        start_microseconds: Option<DateTime<Local>>,
        end_microseconds: Option<DateTime<Local>>,
        selected_host: Option<&String>,
    ) -> Result<Columns> {
        let dbtable = self.get_log_table_name("trace_log");
        let host_filter = self.get_log_host_filter_clause(selected_host);
        return self
            .execute(&format!(
                r#"
            WITH
                {} AS start_time_,
                {} AS end_time_
            SELECT
              {} AS human_trace,
              {} weight
            FROM {}
            WHERE
                    event_date >= toDate(start_time_) AND event_time >= toDateTime(start_time_) AND event_time_microseconds > start_time_
                AND event_date <= toDate(end_time_)   AND event_time <= toDateTime(end_time_)   AND event_time_microseconds <= end_time_
                AND trace_type = '{:?}'
                {}
                {}
            GROUP BY human_trace
            SETTINGS allow_introspection_functions=1
            "#,
                match start_microseconds {
                    Some(time) => format!(
                        "fromUnixTimestamp64Nano({})",
                        time.timestamp_nanos_opt()
                            .ok_or(Error::msg("Invalid start time"))?
                    ),
                    None => "toDateTime64(now() - INTERVAL 1 HOUR, 6)".to_string(),
                },
                match end_microseconds {
                    Some(time) => format!(
                        "fromUnixTimestamp64Nano({})",
                        time.timestamp_nanos_opt()
                            .ok_or(Error::msg("Invalid end time"))?
                    ),
                    None => "toDateTime64(now(), 6)".to_string(),
                },
                if self.quirks.has(ClickHouseAvailableQuirks::TraceLogHasSymbols) {
                    r#"
                        if(empty(symbols),
                           arrayStringConcat(arrayMap(
                             addr -> demangle(addressToSymbol(addr)),
                             arrayReverse(trace)
                           ), ';'),
                           arrayStringConcat(arrayReverse(symbols), ';')
                        )
                    "#
                } else {
                    r#"
                        arrayStringConcat(arrayMap(
                          addr -> demangle(addressToSymbol(addr)),
                          arrayReverse(trace)
                        ), ';')
                    "#
                },
                match trace_type {
                    TraceType::Memory => "abs(sum(size))",
                    TraceType::MemorySample => "abs(sum(size))",
                    TraceType::JemallocSample => "abs(sum(size))",
                    TraceType::MemoryAllocatedWithoutCheck => "abs(sum(size))",
                    _ => "count()",
                },
                dbtable,
                trace_type,
                if let Some(ids) = query_ids {
                    format!("AND query_id IN ('{}')", ids.join("','"))
                } else {
                    String::new()
                },
                host_filter,
            ))
            .await;
    }

    /// Return jemalloc flamegraph in pyspy format.
    /// It is the same format as TSV, but with ' ' delimiter between symbols and weight.
    pub async fn get_jemalloc_flamegraph(&self, selected_host: Option<&String>) -> Result<Columns> {
        let dbtable = self.get_table_name("jemalloc_profile_text");
        let host_filter = if let Some(host) = selected_host {
            if !host.is_empty() && self.options.cluster.is_some() {
                format!("AND hostName() = '{}'", host.replace('\'', "''"))
            } else {
                String::new()
            }
        } else {
            String::new()
        };
        return self
            .execute(&format!(
                r#"
            WITH splitByChar(' ', line) AS parts
            SELECT
                arrayStringConcat(arraySlice(parts, 1, -1), ' ') AS symbols,
                parts[-1]::UInt64 AS bytes
            FROM {}
            WHERE 1 {}
            SETTINGS jemalloc_profile_text_output_format='collapsed'
            "#,
                dbtable, host_filter,
            ))
            .await;
    }

    pub async fn get_live_query_flamegraph(
        &self,
        query_ids: &Option<Vec<String>>,
        selected_host: Option<&String>,
    ) -> Result<Columns> {
        let dbtable = self.get_table_name_no_history("stack_trace");
        let host_filter = self.get_host_filter_clause(selected_host);
        let where_clause = match (query_ids.as_ref(), host_filter.is_empty()) {
            (Some(v), true) => format!("query_id IN ('{}')", v.join("','")),
            (Some(v), false) => format!("query_id IN ('{}') {}", v.join("','"), host_filter),
            (None, false) => format!("1 {}", host_filter),
            (None, true) => "1".to_string(),
        };
        return self
            .execute(&format!(
                r#"
            SELECT
              arrayStringConcat(arrayMap(
                addr -> demangle(addressToSymbol(addr)),
                arrayReverse(trace)
              ), ';') AS human_trace,
              count() weight
            FROM {}
            WHERE {}
            GROUP BY human_trace
            SETTINGS allow_introspection_functions=1
            "#,
                dbtable, where_clause
            ))
            .await;
    }

    pub async fn get_background_schedule_pool_query_ids(
        &self,
        log_name: Option<String>,
        database: String,
        table: String,
        start: RelativeDateTime,
        end: RelativeDateTime,
        selected_host: Option<&String>,
    ) -> Result<Vec<String>> {
        let dbtable = self.get_log_table_name("background_schedule_pool_log");

        let start_sql = start
            .to_sql_datetime_64()
            .ok_or_else(|| Error::msg("Invalid start"))?;
        let end_sql = end
            .to_sql_datetime_64()
            .ok_or_else(|| Error::msg("Invalid end"))?;

        let host_filter = self.get_log_host_filter_clause(selected_host);

        let query = if let Some(ref log_name) = log_name {
            format!(
                r#"
                WITH {start} AS start_, {end} AS end_
                SELECT DISTINCT query_id
                FROM {dbtable}
                WHERE
                    event_date BETWEEN toDate(start_) AND toDate(end_) AND
                    event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                    log_name = '{log_name}' AND
                    database = '{database}' AND
                    table = '{table}'
                    {host_filter}
                LIMIT 1000
                "#,
                start = start_sql,
                end = end_sql,
                dbtable = dbtable,
                log_name = log_name.replace('\'', "''"),
                database = database.replace('\'', "''"),
                table = table.replace('\'', "''"),
                host_filter = host_filter,
            )
        } else {
            format!(
                r#"
                WITH {start} AS start_, {end} AS end_
                SELECT DISTINCT query_id
                FROM {dbtable}
                WHERE
                    event_date BETWEEN toDate(start_) AND toDate(end_) AND
                    event_time BETWEEN toDateTime(start_) AND toDateTime(end_) AND
                    database = '{database}' AND
                    table = '{table}'
                    {host_filter}
                LIMIT 1000
                "#,
                start = start_sql,
                end = end_sql,
                dbtable = dbtable,
                database = database.replace('\'', "''"),
                table = table.replace('\'', "''"),
                host_filter = host_filter,
            )
        };

        let columns = self.execute(&query).await?;
        let mut query_ids = Vec::new();
        for i in 0..columns.row_count() {
            if let Ok(query_id) = columns.get::<String, _>(i, "query_id") {
                query_ids.push(query_id);
            }
        }

        Ok(query_ids)
    }

    pub async fn otel_spans_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("opentelemetry_span_log");
        let start_us = start.timestamp_micros();
        let end_us = end.timestamp_micros();
        let query_id_filter = if let Some(ids) = query_ids {
            format!(
                "AND attribute['clickhouse.query_id'] IN ('{}')",
                ids.join("','")
            )
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    SELECT
                        operation_name,
                        start_time_us,
                        finish_time_us,
                        attribute['clickhouse.query_id'] AS query_id,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    WHERE start_time_us BETWEEN {start_us} AND {end_us}
                      {query_id_filter}
                    ORDER BY start_time_us
                    "#,
            dbtable = dbtable,
            start_us = start_us,
            end_us = end_us,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn trace_log_counters_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("trace_log");
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        query_id,
                        event,
                        increment,
                        event_time_microseconds,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    WHERE trace_type = 'ProfileEvent' AND increment != 0
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn query_metric_log_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("query_metric_log");
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        query_id,
                        event_time_microseconds,
                        memory_usage::Int64 AS memory_usage,
                        peak_memory_usage::Int64 AS peak_memory_usage,
                        {host_expr} AS host_name,
                        COLUMNS('ProfileEvent_')
                    FROM {dbtable}
                    WHERE 1
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn part_log_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("part_log");
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        event_type,
                        event_time_microseconds,
                        duration_ms,
                        database,
                        table,
                        part_name,
                        query_id,
                        rows,
                        size_in_bytes,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    WHERE event_type NOT IN ('MergePartsStart', 'MutatePartStart')
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each(&sql, on_block).await
    }

    fn stack_trace_with_filter(
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
    ) -> Result<(String, String)> {
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let with = format!(
            r#"WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_"#,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        let filter = format!(
            r#"WHERE trace_type IN ('CPU', 'Real', 'Memory')
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)"#,
        );
        Ok((with, filter))
    }

    /// Samples carry only cityHash64(trace) per row; the symbolized stack of
    /// each unique trace comes from stack_traces_for_perfetto_sql() (the dedup
    /// factor is ~100x, transferring stacks per sample is GBs of strings).
    pub async fn stack_trace_samples_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("trace_log");
        let host_expr = self.get_log_hostname_column();
        let (with, filter) = Self::stack_trace_with_filter(query_ids, start, end)?;
        let sql = format!(
            r#"
                    {with}
                    SELECT
                        event_time_microseconds,
                        trace_type,
                        cityHash64(trace) AS stack_hash,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    {filter}
                    ORDER BY event_time_microseconds
                    "#,
        );
        self.execute_for_each(&sql, on_block).await
    }

    /// Maps (host_name, stack_hash) to the symbolized stack of each unique
    /// trace (keyed by host since the same addresses may symbolize
    /// differently across binaries).
    pub async fn stack_traces_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("trace_log");
        let host_expr = self.get_log_hostname_column();
        let symbol_expr = if self
            .quirks
            .has(ClickHouseAvailableQuirks::TraceLogHasSymbols)
        {
            r#"arrayReverse(if(empty(any(symbols)),
                arrayMap(addr -> demangle(addressToSymbol(addr)), trace),
                any(symbols)))"#
        } else {
            "arrayReverse(arrayMap(addr -> demangle(addressToSymbol(addr)), trace))"
        };
        let (with, filter) = Self::stack_trace_with_filter(query_ids, start, end)?;
        let sql = format!(
            r#"
                    {with}
                    SELECT
                        {host_expr} AS host_name,
                        cityHash64(trace) AS stack_hash,
                        {symbol_expr} AS stack
                    FROM {dbtable}
                    {filter}
                    GROUP BY host_name, trace
                    SETTINGS allow_introspection_functions=1
                    "#,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn text_log_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("text_log");
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        event_time_microseconds,
                        level,
                        logger_name,
                        message,
                        query_id,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    WHERE 1
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each_raw(&sql, on_block).await
    }

    pub async fn query_thread_log_for_perfetto(
        &self,
        query_ids: Option<&[String]>,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("query_thread_log");
        let query_id_filter = if let Some(ids) = query_ids {
            format!("AND query_id IN ('{}')", ids.join("','"))
        } else {
            String::new()
        };
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        query_id,
                        thread_name,
                        event_time_microseconds,
                        query_duration_ms,
                        -- Most of ~2K ProfileEvents are zero per thread and Names repeat
                        -- in every row - do not transfer them (GBs for busy servers)
                        arrayFilter((n, v) -> v != 0, mapKeys(ProfileEvents), mapValues(ProfileEvents))
                            AS profile_event_names,
                        arrayFilter(v -> v != 0, mapValues(ProfileEvents))
                            AS profile_event_values,
                        peak_memory_usage,
                        {host_expr} AS host_name
                    FROM {dbtable}
                    WHERE 1
                      {query_id_filter}
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            query_id_filter = query_id_filter,
            host_expr = self.get_log_hostname_column(),
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn queries_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        query_ids: &Option<Vec<String>>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("query_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        -- Query::from_clickhouse_block() compatibility, the perfetto
                        -- export does not use them (and they are heavy to transfer)
                        []::Array(String) AS `ProfileEvents.Names`,
                        []::Array(UInt64) AS `ProfileEvents.Values`,
                        []::Array(String) AS `Settings.Names`,
                        []::Array(String) AS `Settings.Values`,
                        {peak_threads_usage} AS peak_threads_usage,
                        memory_usage::Int64 AS peak_memory_usage,
                        query_duration_ms/1e3 AS elapsed,
                        user,
                        initial_user,
                        exception,
                        is_initial_query,
                        (exception_code = 394)::UInt8 AS is_cancelled,
                        initial_query_id,
                        query_id,
                        hostname as host_name,
                        current_database,
                        query_start_time_microseconds,
                        event_time_microseconds AS query_end_time_microseconds,
                        toValidUTF8(query) AS original_query,
                        normalizeQuery(query) AS normalized_query,
                        normalized_query_hash
                    FROM {dbtable}
                    WHERE type != 'QueryStart'
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                      {query_ids}
                    "#,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
            dbtable = dbtable,
            peak_threads_usage = if self
                .quirks
                .has(ClickHouseAvailableQuirks::QueryLogPeakThreadsUsage)
            {
                "peak_threads_usage"
            } else {
                "length(thread_ids)"
            },
            query_ids = if let Some(query_id) = query_ids {
                format!("AND query_id IN ('{}')", query_id.join("','"))
            } else {
                String::new()
            },
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn get_perfetto_query_scope(
        &self,
        query_id: &str,
        start: DateTime<Local>,
        end: DateTime<Local>,
    ) -> Result<PerfettoQueryScope> {
        let query_log = self.get_log_table_name("query_log");
        let query_id_escaped = query_id.replace('\'', "''");
        let block = self
            .execute(
                format!(
                    r#"
                    WITH
                        '{query_id}' AS root_query_id,
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        groupUniqArray(query_id) AS query_ids,
                        min(query_start_time_microseconds) AS query_start_time_microseconds,
                        max(event_time_microseconds) AS query_end_time_microseconds
                    FROM {query_log}
                    WHERE
                        type != 'QueryStart'
                        AND (query_id = root_query_id OR initial_query_id = root_query_id)
                        AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                        AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    "#,
                    query_id = query_id_escaped,
                    query_log = query_log,
                    start = start
                        .timestamp_nanos_opt()
                        .ok_or(Error::msg("Invalid start"))?,
                    end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
                )
                .as_str(),
            )
            .await?;

        return Ok(PerfettoQueryScope {
            query_ids: Some(block.get::<Vec<String>, _>(0, "query_ids")?),
            start: block
                .get::<DateTime<Tz>, _>(0, "query_start_time_microseconds")?
                .with_timezone(&Local),
            end: block
                .get::<DateTime<Tz>, _>(0, "query_end_time_microseconds")?
                .with_timezone(&Local),
        });
    }

    pub async fn metric_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("metric_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        event_time_microseconds,
                        COLUMNS('ProfileEvent_'),
                        COLUMNS('CurrentMetric_')
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn asynchronous_metric_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("asynchronous_metric_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        metric,
                        value,
                        -- event_time_microseconds was removed from asynchronous_metric_log
                        event_time::DateTime64(6) AS event_time_microseconds
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn asynchronous_insert_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("asynchronous_insert_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        database,
                        table,
                        format,
                        status,
                        bytes,
                        exception,
                        event_time_microseconds,
                        flush_time_microseconds,
                        query_id
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn error_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("error_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        error,
                        code,
                        value,
                        remote,
                        last_error_message,
                        event_time
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn s3_queue_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("s3queue_log");
        let sql = format!(
            r#"
                    SELECT
                        file_name,
                        rows_processed,
                        status,
                        processing_start_time,
                        processing_end_time,
                        exception
                    FROM {dbtable}
                    WHERE processing_start_time >= toDateTime(fromUnixTimestamp64Nano({start}))
                      AND processing_start_time <= toDateTime(fromUnixTimestamp64Nano({end}))
                    ORDER BY processing_start_time
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn azure_queue_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("azure_queue_log");
        let sql = format!(
            r#"
                    SELECT
                        database,
                        table,
                        file_name,
                        rows_processed,
                        status,
                        processing_start_time,
                        processing_end_time,
                        exception
                    FROM {dbtable}
                    WHERE processing_start_time >= toDateTime(fromUnixTimestamp64Nano({start}))
                      AND processing_start_time <= toDateTime(fromUnixTimestamp64Nano({end}))
                    ORDER BY processing_start_time
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn blob_storage_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("blob_storage_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        event_type,
                        query_id,
                        disk_name,
                        bucket,
                        remote_path,
                        data_size,
                        error,
                        event_time_microseconds
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn background_schedule_pool_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("background_schedule_pool_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        log_name,
                        database,
                        table,
                        query_id,
                        duration_ms,
                        error,
                        exception,
                        event_time_microseconds
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn session_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("session_log");
        let sql = format!(
            r#"
                    WITH
                        fromUnixTimestamp64Nano({start}) AS start_,
                        fromUnixTimestamp64Nano({end}) AS end_
                    SELECT
                        type,
                        -- user/auth_type are Nullable
                        coalesce(user, '') AS user,
                        coalesce(auth_type, '') AS auth_type,
                        interface,
                        toString(client_address) AS client_address,
                        client_name,
                        failure_reason,
                        event_time_microseconds
                    FROM {dbtable}
                    WHERE 1
                      AND event_date >= toDate(start_) AND event_time >= toDateTime(start_)
                      AND event_date <= toDate(end_)   AND event_time <= toDateTime(end_)
                    ORDER BY event_time_microseconds
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn aggregated_zookeeper_log_for_perfetto(
        &self,
        start: DateTime<Local>,
        end: DateTime<Local>,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let dbtable = self.get_log_table_name("aggregated_zookeeper_log");
        let sql = format!(
            r#"
                    SELECT
                        event_time,
                        session_id,
                        parent_path,
                        operation,
                        count::UInt64 AS count,
                        mapKeys(errors) AS error_names,
                        mapValues(errors) AS error_counts,
                        average_latency,
                        component
                    FROM {dbtable}
                    WHERE event_time >= toDateTime(fromUnixTimestamp64Nano({start}))
                      AND event_time <= toDateTime(fromUnixTimestamp64Nano({end}))
                    ORDER BY event_time
                    "#,
            dbtable = dbtable,
            start = start
                .timestamp_nanos_opt()
                .ok_or(Error::msg("Invalid start"))?,
            end = end.timestamp_nanos_opt().ok_or(Error::msg("Invalid end"))?,
        );
        self.execute_for_each(&sql, on_block).await
    }

    pub async fn get_warnings(&self) -> Result<Vec<String>> {
        let table_exists: u64 = self
            .execute(
                "SELECT count() FROM system.tables WHERE database = 'system' AND name = 'warnings'",
            )
            .await?
            .get(0, "count()")?;
        if table_exists == 0 {
            return Ok(Vec::new());
        }

        let block = self.execute("SELECT message FROM system.warnings").await?;
        let warnings: Vec<String> = collect_values(&block, "message");
        let filtered: Vec<String> = warnings
            .into_iter()
            .filter(|w| !w.contains("transparent_hugepage") && !w.starts_with("Obsolete settings"))
            .collect();
        Ok(filtered)
    }

    pub async fn execute(&self, query: impl Into<Query>) -> Result<Columns> {
        let query = query.into();
        let columns = self
            .pool
            .get_handle()
            .await?
            .query(query.clone())
            .fetch_all()
            .await?;
        log::trace!(
            "Received {} rows for query: {:?}",
            columns.row_count(),
            query
        );
        Ok(columns)
    }

    /// Stream the result block-by-block, so the whole result is never
    /// materialized in memory and a slow consumer backpressures the server
    /// (the stream is not polled while `on_block` is pending). `on_block`
    /// returning false stops the query early.
    pub async fn execute_for_each(
        &self,
        query: &str,
        on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        // Blocks are capped by rows, not bytes: with wide rows (e.g. ~2K
        // metric_log columns) a default 65K-row block is GBs of allocations,
        // and consumers keep a few blocks in flight. A per-query setting rather
        // than a SETTINGS clause, so it composes with queries that already carry
        // their own SETTINGS (e.g. allow_introspection_functions).
        let query = Query::new(query).with_setting("max_block_size", 8192u64, false);
        self.execute_for_each_raw(query, on_block).await
    }

    /// Same as execute_for_each(), but keeps the server default block size —
    /// for narrow-row tables (e.g. text_log) where capping it only slows the
    /// scan down.
    async fn execute_for_each_raw(
        &self,
        query: impl Into<Query>,
        mut on_block: impl AsyncFnMut(Block) -> bool,
    ) -> Result<()> {
        let query = query.into();
        let mut client = self.pool.get_handle().await?;
        let mut stream = client.query(query.clone()).stream_blocks();
        let mut rows = 0;
        while let Some(block) = stream.next().await {
            let block = block?;
            if block.row_count() == 0 {
                continue;
            }
            rows += block.row_count();
            if !on_block(block).await {
                break;
            }
        }
        log::trace!("Received {} rows (streamed) for query: {:?}", rows, query);
        Ok(())
    }

    async fn execute_simple(&self, query: &str) -> Result<()> {
        let mut client = self.pool.get_handle().await?;
        let mut stream = client.query(query).stream_blocks();
        let ret = stream.next().await;
        if let Some(Err(err)) = ret {
            return Err(Error::new(err));
        } else {
            return Ok(());
        }
    }

    pub async fn get_cluster_hosts(&self) -> Result<Vec<String>> {
        let cluster = self.options.cluster.clone().unwrap_or_default();
        if cluster.is_empty() {
            return Ok(Vec::new());
        }

        let query = format!(
            "SELECT DISTINCT hostName() AS host FROM clusterAllReplicas('{}', system.one) ORDER BY host",
            cluster
        );

        let columns = self.execute(&query).await?;
        let mut hosts = Vec::new();
        for i in 0..columns.row_count() {
            if let Ok(host) = columns.get::<String, _>(i, "host") {
                hosts.push(host);
            }
        }

        Ok(hosts)
    }

    pub fn get_host_filter_clause(&self, selected_host: Option<&String>) -> String {
        if let Some(host) = selected_host
            && !host.is_empty()
            && self.options.cluster.is_some()
        {
            return format!("AND hostName() = '{}'", host.replace('\'', "''"));
        }
        String::new()
    }

    // Filter for system.*_log reads. Without clusterAllReplicas(), hostName() collapses to the
    // executor node, so we match on the persisted `hostname` column instead.
    pub fn get_log_host_filter_clause(&self, selected_host: Option<&String>) -> String {
        if let Some(host) = selected_host
            && !host.is_empty()
            && self.options.cluster.is_some()
        {
            let col = if self.shared_log_pipeline {
                "hostname"
            } else {
                "hostName()"
            };
            return format!("AND {} = '{}'", col, host.replace('\'', "''"));
        }
        String::new()
    }

    pub fn get_internal_filter_clause(&self) -> String {
        if self.options.internal_queries {
            String::new()
        } else {
            format!("AND client_name != '{}'", get_client_name())
        }
    }

    // SELECT-side hostname expression for system.*_log reads. Pairs with get_log_host_filter_clause.
    pub fn get_log_hostname_column(&self) -> &'static str {
        if self.shared_log_pipeline {
            "hostname"
        } else {
            "hostName()"
        }
    }

    /// Database with the system tables ("system" unless overridden with --database).
    pub fn system_database(&self) -> &str {
        self.options.database.as_deref().unwrap_or("system")
    }

    pub fn get_table_name(&self, table: &str) -> String {
        let database = self.system_database();
        let cluster = self.options.cluster.clone().unwrap_or_default();
        let history = self.options.history;

        return match (history, cluster.is_empty()) {
            (false, true) => format!("{}.{}", database, table),
            (true, false) => format!(
                "clusterAllReplicas('{}', merge('{}', '^{}'))",
                cluster, database, table
            ),
            (true, true) => format!("merge('{}', '^{}')", database, table),
            (false, false) => format!(
                "clusterAllReplicas('{}', '{}', '{}')",
                cluster, database, table
            ),
        };
    }

    // Variant for system.*_log tables. With use_shared_merge_tree_log_pipeline we can skip
    // clusterAllReplicas() entirely — a single replica observes the whole cluster's rows.
    pub fn get_log_table_name(&self, table: &str) -> String {
        if self.shared_log_pipeline {
            let database = self.system_database();
            let history = self.options.history;
            return if history {
                format!("merge('{}', '^{}')", database, table)
            } else {
                format!("{}.{}", database, table)
            };
        }
        self.get_table_name(table)
    }

    pub fn get_table_name_no_history(&self, table: &str) -> String {
        self.get_table_name_no_history_in(self.system_database(), table)
    }

    /// For tables with live server state (e.g. system.processes): a preserved
    /// copy of the system tables (--database) cannot contain it.
    pub fn get_live_table_name(&self, table: &str) -> String {
        self.get_table_name_no_history_in("system", table)
    }

    fn get_table_name_no_history_in(&self, database: &str, table: &str) -> String {
        let cluster = self.options.cluster.clone().unwrap_or_default();
        return match cluster.is_empty() {
            true => format!("{}.{}", database, table),
            false => format!(
                "clusterAllReplicas('{}', '{}', '{}')",
                cluster, database, table
            ),
        };
    }
}