refactor: remove redundant sql from result cache meta (#19736)

* feat: limit result cache by sql length

* refactor: store result cache sql hash in meta

* refactor: remove redundant sql from result cache meta

* refactor: show query id in result cache explain

* refactor: keep result cache explain stable

* refactor: log query id for result cache access

* fix: resolve result cache ci failures

* fix(storage): reorder basic Cargo deps

* test: update result cache explain sqllogic outputs

* test: remove sql from result cache explain outputs

Author: sundy-li

Cargo.lock

@@ -364,7 +364,6 @@ impl ExplainInterpreter {
             if let Some(v) = cache_reader.check_cache().await? {
                 // Construct a format tree for result cache reading
                 let children = vec![
-                    FormatTreeNode::new(format!("SQL: {}", v.sql)),
                     FormatTreeNode::new(format!("Number of rows: {}", v.num_rows)),
                     FormatTreeNode::new(format!("Result size: {}", v.result_size)),
                 ];

src/query/service/src/interpreters/interpreter_explain.rs

@@ -394,7 +394,6 @@ DB.Table: 'system'.'columns', Table: columns-table_id:1, ver:0, Engine: SystemCo
 | 'snapshot_location'               | 'system'             | 'streams'                 | 'Nullable(String)'    | 'VARCHAR'           | ''       | ''       | 'YES'    | ''       |
 | 'source'                          | 'system'             | 'dictionaries'            | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'source_column'                   | 'system'             | 'virtual_columns'         | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
-| 'sql'                             | 'system'             | 'query_cache'             | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'sql_path'                        | 'information_schema' | 'schemata'                | 'NULL'                | 'NULL'              | ''       | ''       | 'NO'     | ''       |
 | 'stack'                           | 'system'             | 'backtrace'               | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'stage_type'                      | 'system'             | 'stages'                  | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |

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

@@ -22,6 +22,7 @@ databend-common-storages-parquet = { workspace = true }
 databend-meta-client = { workspace = true }
 databend-storages-common-blocks = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }
+log = { workspace = true }
 opendal = { workspace = true }
 parking_lot = { workspace = true }
 parquet = { workspace = true }

src/query/storages/basic/Cargo.toml

@@ -39,8 +39,6 @@ pub(crate) fn gen_result_cache_dir(key: &str) -> String {
 
 #[derive(serde::Serialize, serde::Deserialize)]
 pub struct ResultCacheValue {
-    /// The original query SQL.
-    pub sql: String,
     /// Associated query id
     pub query_id: String,
     /// The query time.

src/query/storages/basic/src/result_cache/common.rs

@@ -20,6 +20,7 @@ use databend_common_expression::DataBlock;
 use databend_common_expression::DataSchema;
 use databend_common_meta_store::MetaStore;
 use databend_common_storage::DataOperator;
+use log::info;
 use opendal::Operator;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReader;
 
@@ -68,6 +69,10 @@ impl ResultCacheReader {
     pub async fn check_cache(&self) -> Result<Option<ResultCacheValue>> {
         if let Some(v) = self.meta_mgr.get(self.meta_key.clone()).await? {
             if self.tolerate_inconsistent || v.partitions_shas == self.partitions_shas {
+                info!(
+                    "Query result cache hit: query_id={}, meta_key={}",
+                    v.query_id, self.meta_key
+                );
                 return Ok(Some(v));
             }
         }

src/query/storages/basic/src/result_cache/read/reader.rs

@@ -27,6 +27,7 @@ use databend_common_pipeline::sinks::AsyncMpscSink;
 use databend_common_pipeline::sinks::AsyncMpscSinker;
 use databend_common_storage::DataOperator;
 use databend_meta_client::types::MatchSeq;
+use log::info;
 use tokio::time::Instant;
 
 use super::writer::ResultCacheWriter;
@@ -37,7 +38,6 @@ use crate::result_cache::meta_manager::ResultCacheMetaManager;
 
 pub struct WriteResultCacheSink {
     ctx: Arc<dyn TableContext>,
-    sql: String,
     partitions_shas: Vec<String>,
 
     meta_mgr: ResultCacheMetaManager,
@@ -109,7 +109,6 @@ impl AsyncMpscSink for WriteResultCacheSink {
         let ttl_interval = Duration::from_secs(ttl_sec);
 
         let value = ResultCacheValue {
-            sql: self.sql.clone(),
             query_id: self.ctx.get_id(),
             query_time: now,
             ttl: ttl_sec,
@@ -122,6 +121,11 @@ impl AsyncMpscSink for WriteResultCacheSink {
         self.meta_mgr
             .set(self.meta_key.clone(), value, MatchSeq::GE(0), ttl_interval)
             .await?;
+        info!(
+            "Query result cache write: query_id={}, meta_key={}",
+            self.ctx.get_id(),
+            self.meta_key
+        );
         self.ctx
             .set_query_id_result_cache(self.ctx.get_id(), self.meta_key.clone());
         Ok(())
@@ -141,7 +145,6 @@ impl WriteResultCacheSink {
         let min_execute_secs = settings.get_query_result_cache_min_execute_secs()?;
         let ttl = settings.get_query_result_cache_ttl_secs()?;
         let tenant = ctx.get_tenant();
-        let sql = ctx.get_query_str();
         let partitions_shas = ctx.get_partitions_shas();
 
         let meta_key = gen_result_cache_meta_key(tenant.tenant_name(), key);
@@ -155,7 +158,6 @@ impl WriteResultCacheSink {
             inputs,
             WriteResultCacheSink {
                 ctx,
-                sql,
                 partitions_shas,
                 meta_mgr: ResultCacheMetaManager::create(kv_store, ttl),
                 meta_key,

src/query/storages/basic/src/result_cache/write/sink.rs

@@ -63,7 +63,6 @@ impl AsyncSystemTable for QueryCacheTable {
 
         let cached_values = result_cache_mgr.list(prefix.as_str()).await?;
 
-        let mut sql_vec: Vec<&str> = Vec::with_capacity(cached_values.len());
         let mut query_id_vec: Vec<&str> = Vec::with_capacity(cached_values.len());
         let mut result_size_vec = Vec::with_capacity(cached_values.len());
         let mut num_rows_vec = Vec::with_capacity(cached_values.len());
@@ -73,7 +72,6 @@ impl AsyncSystemTable for QueryCacheTable {
         let mut active_result_scan: Vec<bool> = Vec::with_capacity(cached_values.len());
 
         cached_values.iter().for_each(|x| {
-            sql_vec.push(x.sql.as_str());
             query_id_vec.push(x.query_id.as_str());
             result_size_vec.push(x.result_size as u64);
             num_rows_vec.push(x.num_rows as u64);
@@ -103,7 +101,6 @@ impl AsyncSystemTable for QueryCacheTable {
             .collect();
 
         Ok(DataBlock::new_from_columns(vec![
-            StringType::from_data(sql_vec),
             StringType::from_data(query_id_vec),
             UInt64Type::from_data(result_size_vec),
             UInt64Type::from_data(num_rows_vec),
@@ -128,7 +125,6 @@ impl AsyncSystemTable for QueryCacheTable {
 impl QueryCacheTable {
     pub fn create(table_id: u64) -> Arc<dyn Table> {
         let schema = TableSchemaRefExt::create(vec![
-            TableField::new("sql", TableDataType::String),
             TableField::new("query_id", TableDataType::String),
             TableField::new("result_size", TableDataType::Number(NumberDataType::UInt64)),
             TableField::new("num_rows", TableDataType::Number(NumberDataType::UInt64)),

src/query/storages/system/src/query_cache_table.rs

@@ -79,15 +79,13 @@ query T
 EXPLAIN SELECT * FROM t1 ORDER BY a;
 ----
 ReadQueryResultCache
-├── SQL: SELECT * FROM t1 ORDER BY a
 ├── Number of rows: 3
 └── Result size: 12
 
 query T
 EXPLAIN SELECT * FROM t1, t2 ORDER BY a, b;
 ----
 ReadQueryResultCache
-├── SQL: SELECT * FROM t1, t2 ORDER BY a, b
 ├── Number of rows: 9
 └── Result size: 180
 
@@ -320,7 +318,6 @@ query T
 EXPLAIN SELECT *, (SELECT MIN(b) FROM t_scalar_subquery) as min_val FROM t_scalar_subquery ORDER BY a;
 ----
 ReadQueryResultCache
-├── SQL: SELECT *, (SELECT MIN(b) FROM t_scalar_subquery) AS min_val FROM t_scalar_subquery ORDER BY a
 ├── Number of rows: 3
 └── Result size: 39
 

tests/sqllogictests/suites/base/20+_others/20_0013_query_result_cache.test

@@ -72,7 +72,6 @@ query T
 EXPLAIN SELECT * FROM rap_cache_test ORDER BY id;
 ----
 ReadQueryResultCache
-├── SQL: SELECT * FROM rap_cache_test ORDER BY id
 ├── Number of rows: 2
 └── Result size: 119
 
@@ -126,7 +125,6 @@ query T
 EXPLAIN SELECT * FROM rap_cache_test ORDER BY id;
 ----
 ReadQueryResultCache
-├── SQL: SELECT * FROM rap_cache_test ORDER BY id
 ├── Number of rows: 3
 └── Result size: 177
 
@@ -142,7 +140,6 @@ query T
 EXPLAIN SELECT * FROM rap_cache_test ORDER BY id;
 ----
 ReadQueryResultCache
-├── SQL: SELECT * FROM rap_cache_test ORDER BY id
 ├── Number of rows: 2
 └── Result size: 119