refactor: query_history table should only have one record per query (#18098)

* use MERGE instead of INSERT

* error retry should have limited times

* add config `log_only` into HistoryConfig

Author: dqhl76

src/binaries/query/entry.rs

@@ -291,7 +291,7 @@ pub async fn start_services(conf: &InnerConfig) -> Result<(), MainError> {
     if conf.log.structlog.on {
         println!("    structlog: {}", conf.log.structlog);
     }
-    if conf.log.history.on {
+    if conf.log.history.on && !conf.log.history.log_only {
         GlobalHistoryLog::instance().initialized();
         println!("    system history tables: {}", conf.log.history);
     }

src/common/tracing/src/config.rs

@@ -340,6 +340,7 @@ impl Default for OTLPEndpointConfig {
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
 pub struct HistoryConfig {
     pub on: bool,
+    pub log_only: bool,
     pub interval: usize,
     pub stage_name: String,
     pub level: String,
@@ -357,8 +358,9 @@ impl Display for HistoryConfig {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(
             f,
-            "enabled={}, interval={}, stage_name={}, level={}, retention_interval={}, tables=[{}]",
+            "enabled={}, log_only={}, interval={}, stage_name={}, level={}, retention_interval={}, tables=[{}]",
             self.on,
+            self.log_only,
             self.interval,
             self.stage_name,
             self.level,
@@ -376,6 +378,7 @@ impl Default for HistoryConfig {
         Self {
             on: false,
             interval: 2,
+            log_only: false,
             // The default value of stage name uses an uuid to avoid conflicts with existing stages
             stage_name: "log_1f93b76af0bd4b1d8e018667865fbc65".to_string(),
             level: "WARN".to_string(),

src/common/tracing/src/predefined_tables/history_tables.toml

@@ -10,7 +10,7 @@ delete = "DELETE FROM system_history.log_history WHERE timestamp < subtract_hour
 name = "query_history"
 target = "databend::log::query"
 create = "CREATE TABLE IF NOT EXISTS system_history.query_history (log_type Int8 NULL, log_type_name STRING NULL, handler_type STRING NULL, tenant_id STRING NULL, cluster_id STRING NULL, node_id STRING NULL, sql_user STRING NULL, sql_user_quota STRING NULL, sql_user_privileges STRING NULL, query_id STRING NULL, query_kind STRING NULL, query_text STRING NULL, query_hash STRING NULL, query_parameterized_hash STRING NULL, event_date DATE NULL, event_time TIMESTAMP NULL, query_start_time TIMESTAMP NULL, query_duration_ms Int64 NULL, query_queued_duration_ms Int64 NULL, current_database STRING NULL, written_rows UInt64 NULL, written_bytes UInt64 NULL, join_spilled_rows UInt64 NULL, join_spilled_bytes UInt64 NULL, agg_spilled_rows UInt64 NULL, agg_spilled_bytes UInt64 NULL, group_by_spilled_rows UInt64 NULL, group_by_spilled_bytes UInt64 NULL, written_io_bytes UInt64 NULL, written_io_bytes_cost_ms UInt64 NULL, scan_rows UInt64 NULL, scan_bytes UInt64 NULL, scan_io_bytes UInt64 NULL, scan_io_bytes_cost_ms UInt64 NULL, scan_partitions UInt64 NULL, total_partitions UInt64 NULL, result_rows UInt64 NULL, result_bytes UInt64 NULL, bytes_from_remote_disk UInt64 NULL, bytes_from_local_disk UInt64 NULL, bytes_from_memory UInt64 NULL, client_address STRING NULL, user_agent STRING NULL, exception_code Int32 NULL, exception_text STRING NULL, server_version STRING NULL, query_tag STRING NULL, has_profile BOOLEAN NULL, peek_memory_usage VARIANT NULL, session_id STRING NULL) CLUSTER BY LINEAR(event_time, query_id)"
-transform = "settings (timezone='Etc/UTC') INSERT INTO system_history.query_history FROM (SELECT m['log_type'], m['log_type_name'], m['handler_type'], m['tenant_id'], m['cluster_id'], m['node_id'], m['sql_user'], m['sql_user_quota'], m['sql_user_privileges'], m['query_id'], m['query_kind'], m['query_text'], m['query_hash'], m['query_parameterized_hash'], m['event_date'], m['event_time'], m['query_start_time'], m['query_duration_ms'], m['query_queued_duration_ms'], m['current_database'], m['written_rows'], m['written_bytes'], m['join_spilled_rows'], m['join_spilled_bytes'], m['agg_spilled_rows'], m['agg_spilled_bytes'], m['group_by_spilled_rows'], m['group_by_spilled_bytes'], m['written_io_bytes'], m['written_io_bytes_cost_ms'], m['scan_rows'], m['scan_bytes'], m['scan_io_bytes'], m['scan_io_bytes_cost_ms'], m['scan_partitions'], m['total_partitions'], m['result_rows'], m['result_bytes'], m['bytes_from_remote_disk'], m['bytes_from_local_disk'], m['bytes_from_memory'], m['client_address'], m['user_agent'], m['exception_code'], m['exception_text'], m['server_version'], m['query_tag'], m['has_profile'], m['peek_memory_usage'], m['session_id'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::query' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
+transform = "settings (timezone='Etc/UTC') MERGE INTO system_history.query_history AS target USING (SELECT f['log_type'] AS log_type, f['log_type_name'] AS log_type_name, f['handler_type'] AS handler_type, f['tenant_id'] AS tenant_id, f['cluster_id'] AS cluster_id, f['node_id'] AS node_id, f['sql_user'] AS sql_user, f['sql_user_quota'] AS sql_user_quota, f['sql_user_privileges'] AS sql_user_privileges, f['query_id'] AS query_id, f['query_kind'] AS query_kind, f['query_text'] AS query_text, f['query_hash'] AS query_hash, f['query_parameterized_hash'] AS query_parameterized_hash, f['event_date'] AS event_date, f['event_time'] AS event_time, f['query_start_time'] AS query_start_time, f['query_duration_ms'] AS query_duration_ms, f['query_queued_duration_ms'] AS query_queued_duration_ms, f['current_database'] AS current_database, f['written_rows'] AS written_rows, f['written_bytes'] AS written_bytes, f['join_spilled_rows'] AS join_spilled_rows, f['join_spilled_bytes'] AS join_spilled_bytes, f['agg_spilled_rows'] AS agg_spilled_rows, f['agg_spilled_bytes'] AS agg_spilled_bytes, f['group_by_spilled_rows'] AS group_by_spilled_rows, f['group_by_spilled_bytes'] AS group_by_spilled_bytes, f['written_io_bytes'] AS written_io_bytes, f['written_io_bytes_cost_ms'] AS written_io_bytes_cost_ms, f['scan_rows'] AS scan_rows, f['scan_bytes'] AS scan_bytes, f['scan_io_bytes'] AS scan_io_bytes, f['scan_io_bytes_cost_ms'] AS scan_io_bytes_cost_ms, f['scan_partitions'] AS scan_partitions, f['total_partitions'] AS total_partitions, f['result_rows'] AS result_rows, f['result_bytes'] AS result_bytes, f['bytes_from_remote_disk'] AS bytes_from_remote_disk, f['bytes_from_local_disk'] AS bytes_from_local_disk, f['bytes_from_memory'] AS bytes_from_memory, f['client_address'] AS client_address, f['user_agent'] AS user_agent, f['exception_code'] AS exception_code, f['exception_text'] AS exception_text, f['server_version'] AS server_version, f['query_tag'] AS query_tag, f['has_profile'] AS has_profile, f['peek_memory_usage'] AS peek_memory_usage, f['session_id'] AS session_id FROM (SELECT ARG_MAX(m, m['log_type']) AS f FROM (SELECT parse_json(message) AS m FROM system_history.log_history WHERE target = 'databend::log::query' AND batch_number >= {batch_begin} AND batch_number < {batch_end}) AS parsed_data GROUP BY m['query_id'])) AS source ON target.query_id = source.query_id WHEN MATCHED AND source.log_type IN (2, 3) THEN UPDATE * WHEN NOT MATCHED THEN INSERT *;"
 delete = "DELETE FROM system_history.query_history WHERE event_time < subtract_hours(NOW(), {retention_hours})"
 
 [[tables]]

src/query/config/src/config.rs

@@ -2724,6 +2724,21 @@ pub struct HistoryLogConfig {
     #[serde(rename = "on")]
     pub log_history_on: bool,
 
+    /// Enables log-only mode for the history feature.
+    ///
+    /// When set to true, this node will only record raw log data to the specified stage,
+    /// but will not perform the transform and clean process.
+    /// Please make sure that the transform and clean process is handled by other nodes
+    /// otherwise the raw log data will not be processed and cleaned up.
+    ///
+    /// Note: This is useful for nodes that should avoid the performance overhead of the
+    /// transform and clean process
+    #[clap(
+        long = "log-history-log-only", value_name = "VALUE", default_value = "false", action = ArgAction::Set, num_args = 0..=1, require_equals = true, default_missing_value = "true"
+    )]
+    #[serde(rename = "log_only")]
+    pub log_history_log_only: bool,
+
     /// Specifies the interval in seconds for how often the history log is flushed
     #[clap(
         long = "log-history-interval",
@@ -2819,6 +2834,7 @@ impl TryInto<InnerHistoryConfig> for HistoryLogConfig {
     fn try_into(self) -> Result<InnerHistoryConfig> {
         Ok(InnerHistoryConfig {
             on: self.log_history_on,
+            log_only: self.log_history_log_only,
             interval: self.log_history_interval,
             stage_name: self.log_history_stage_name,
             level: self.log_history_level,
@@ -2836,6 +2852,7 @@ impl From<InnerHistoryConfig> for HistoryLogConfig {
     fn from(inner: InnerHistoryConfig) -> Self {
         Self {
             log_history_on: inner.on,
+            log_history_log_only: inner.log_only,
             log_history_interval: inner.interval,
             log_history_stage_name: inner.stage_name,
             log_history_level: inner.level,

src/query/service/src/history_tables/global_history_log.rs

@@ -74,6 +74,12 @@ pub struct GlobalHistoryLog {
 impl GlobalHistoryLog {
     pub async fn init(cfg: &InnerConfig) -> Result<()> {
         setup_operator().await?;
+        if cfg.log.history.log_only {
+            info!(
+                "[HISTORY-TABLES] History tables transform is disabled, only logging is enabled."
+            );
+            return Ok(());
+        }
         let meta_client = MetaGrpcClient::try_new(&cfg.meta.to_meta_grpc_client_conf())
             .map_err(|_e| ErrorCode::Internal("Create MetaClient failed for SystemHistory"))?;
         let stage_name = cfg.log.history.stage_name.clone();
@@ -135,10 +141,12 @@ impl GlobalHistoryLog {
             let meta_key = format!("{}/history_log_transform", self.tenant_id).clone();
             let log = GlobalHistoryLog::instance();
             let handle = spawn(async move {
+                let mut consecutive_error = 0;
                 loop {
                     match log.transform(&table_clone, &meta_key).await {
                         Ok(acquired_lock) => {
                             if acquired_lock {
+                                consecutive_error = 0;
                                 let _ = log
                                     .finish_hook(&format!("{}/{}/lock", meta_key, table_clone.name))
                                     .await;
@@ -148,21 +156,18 @@ impl GlobalHistoryLog {
                             let _ = log
                                 .finish_hook(&format!("{}/{}/lock", meta_key, table_clone.name))
                                 .await;
-
-                            // BadArguments(1006), if the table schema is changed
-                            // means this node is older version then exit
-                            if e.code() == 1006 {
-                                info!(
-                                    "[HISTORY-TABLES] {} log transform failed due to schema changed, exit",
+                            error!(
+                                "[HISTORY-TABLES] {} log transform failed due to {}, retry {}",
+                                table_clone.name, e, consecutive_error
+                            );
+                            consecutive_error += 1;
+                            if consecutive_error > 3 {
+                                error!(
+                                    "[HISTORY-TABLES] {} log transform failed too many times, exit",
                                     table_clone.name
                                 );
                                 break;
                             }
-
-                            error!(
-                                "[HISTORY-TABLES] {} log transform failed due to {}, retry",
-                                table_clone.name, e
-                            );
                         }
                     }
                     sleep(sleep_time).await;

src/query/service/tests/it/storages/testdata/configs_table_basic.txt

@@ -41,6 +41,7 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | 'log'     | 'file.prefix_filter'                            | 'null'                                                                                                                                                                                                    | ''       |
 | 'log'     | 'history.interval'                              | '2'                                                                                                                                                                                                       | ''       |
 | 'log'     | 'history.level'                                 | 'WARN'                                                                                                                                                                                                    | ''       |
+| 'log'     | 'history.log_only'                              | 'false'                                                                                                                                                                                                   | ''       |
 | 'log'     | 'history.on'                                    | 'false'                                                                                                                                                                                                   | ''       |
 | 'log'     | 'history.retention_interval'                    | '24'                                                                                                                                                                                                      | ''       |
 | 'log'     | 'history.stage_name'                            | 'log_1f93b76af0bd4b1d8e018667865fbc65'                                                                                                                                                                    | ''       |

tests/logging/check_logs_table.sh

@@ -40,4 +40,4 @@ check_query_log "2" "$query_id" "SELECT count(*) FROM system_history.profile_his
 check_query_log "3" "$query_id" "SELECT count(*) FROM system_history.log_history WHERE target = 'databend::log::query' and" "2"
 
 # Test 4
-check_query_log "4" "$query_id" "SELECT count(*) FROM system_history.query_history WHERE" "2"
+check_query_log "4" "$query_id" "SELECT count(*) FROM system_history.query_history WHERE" "1"