feat(query): refresh TopN statistics on append (#20091)

* feat(query): refresh topn statistics on append

* refactor(query): simplify append topn stats commit

* fix: add missing topn capacity in selectivity tests

* fix(storage): drop stale CMS on append TopN refresh

* fix(storage): don't seed TopN from missing stats

* fix: drop stale histograms on append topn refresh

* fix(storage): refresh topn for added nullable columns

* feat(fuse): cache TopN statistics per segment

* revert: cache TopN statistics per segment

* feat(fuse): cache per-block TopN in segment statistics

* fix(fuse): address statistics review feedback

* fix(fuse): refresh stats metadata for TopN-only appends

Author: KKould

src/common/frozen_api/src/lib.rs

@@ -234,6 +234,7 @@ fn is_custom_type(name: &str) -> bool {
             | "ClusterKey"
             | "StatisticsOfColumns"
             | "BlockHLL"
+            | "BlockTopN"
             | "SnapshotId"
     )
 }

src/query/expression/src/values.rs

@@ -116,6 +116,8 @@ use crate::with_opaque_size;
 use crate::with_opaque_size_mapped;
 use crate::with_opaque_type;
 
+pub const LARGE_STRING_BYTES_THRESHOLD: usize = 256;
+
 #[derive(Debug, Clone, PartialEq, EnumAsInner)]
 pub enum Value<T: AccessType> {
     Scalar(T::Scalar),
@@ -1913,7 +1915,7 @@ impl Column {
         }
     }
 
-    /// Checks if the average length of a string column exceeds 256 bytes.
+    /// Checks if the average length of a string column exceeds LARGE_STRING_BYTES_THRESHOLD bytes.
     /// If it does, the bloom index for the column will not be established.
     pub fn check_large_string(&self) -> bool {
         let (inner, len) = if let Column::Nullable(c) = self {
@@ -1923,7 +1925,7 @@ impl Column {
         };
         if let Column::String(v) = inner {
             let bytes_per_row = v.total_bytes_len() / len.max(1);
-            if bytes_per_row > 256 {
+            if bytes_per_row > LARGE_STRING_BYTES_THRESHOLD {
                 return true;
             }
         }

src/query/service/src/interpreters/interpreter_table_drop_column.rs

@@ -26,6 +26,7 @@ use databend_common_sql::binder::validate_table_indexes_not_referencing_columns;
 use databend_common_sql::plans::DropTableColumnPlan;
 use databend_common_storages_basic::view_table::VIEW_ENGINE;
 use databend_common_storages_stream::stream_table::STREAM_ENGINE;
+use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_FREQUENCY_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
 
@@ -159,6 +160,16 @@ impl Interpreter for DropTableColumnInterpreter {
                 }
             }
         }
+        if let Some(value) = opts.get_mut(OPT_KEY_ANALYZE_FREQUENCY_COLUMNS) {
+            if let ApproxDistinctColumns::Specify(mut cols) =
+                value.parse::<ApproxDistinctColumns>()?
+            {
+                if let Some(pos) = cols.iter().position(|x| *x == self.plan.column) {
+                    cols.remove(pos);
+                    *value = cols.join(",");
+                }
+            }
+        }
         let new_schema = new_table_meta.schema.as_ref().clone();
 
         let dropped_column_ids = field.column_ids().into_iter().collect();

src/query/service/src/interpreters/interpreter_table_modify_column.rs

@@ -61,6 +61,7 @@ use databend_storages_common_index::RangeIndex;
 use databend_storages_common_table_meta::meta::SnapshotId;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
+use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_FREQUENCY_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
 
@@ -308,6 +309,12 @@ impl ModifyTableColumnInterpreter {
                 approx_distinct_cols = cols;
             }
         }
+        let mut analyze_frequency_cols = vec![];
+        if let Some(v) = table_info.options().get(OPT_KEY_ANALYZE_FREQUENCY_COLUMNS) {
+            if let ApproxDistinctColumns::Specify(cols) = v.parse::<ApproxDistinctColumns>()? {
+                analyze_frequency_cols = cols;
+            }
+        }
 
         let mut table_info = table.get_table_info().clone();
         table_info.meta.fill_field_comments();
@@ -345,6 +352,16 @@ impl ModifyTableColumnInterpreter {
                             field.data_type
                         )));
                     }
+                    if analyze_frequency_cols
+                        .iter()
+                        .any(|v| v.as_str() == field.name)
+                        && !RangeIndex::supported_table_type(&field.data_type)
+                    {
+                        return Err(ErrorCode::TableOptionInvalid(format!(
+                            "Unsupported data type '{}' for analyze frequency columns",
+                            field.data_type
+                        )));
+                    }
                     // If the column is inverted index column, the type can't be changed.
                     if !table_info.meta.indexes.is_empty() {
                         for (index_name, index) in &table_info.meta.indexes {

src/query/service/src/interpreters/interpreter_table_rename_column.rs

@@ -23,6 +23,7 @@ use databend_common_sql::plans::RenameTableColumnPlan;
 use databend_common_storages_basic::view_table::VIEW_ENGINE;
 use databend_common_storages_iceberg::table::ICEBERG_ENGINE;
 use databend_common_storages_stream::stream_table::STREAM_ENGINE;
+use databend_storages_common_table_meta::table::OPT_KEY_ANALYZE_FREQUENCY_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_APPROX_DISTINCT_COLUMNS;
 use databend_storages_common_table_meta::table::OPT_KEY_BLOOM_INDEX_COLUMNS;
 
@@ -129,6 +130,14 @@ impl Interpreter for RenameTableColumnInterpreter {
                 &self.plan.new_column,
             )?;
         }
+        if let Some(value) = opts.get_mut(OPT_KEY_ANALYZE_FREQUENCY_COLUMNS) {
+            rename_column_in_comma_separated_ident(
+                self.ctx.as_ref(),
+                value,
+                &self.plan.old_column,
+                &self.plan.new_column,
+            )?;
+        }
 
         let mut new_cluster_key = None;
         if let Some((_, cluster_key)) = table.cluster_key_meta() {

src/query/service/src/physical_plans/physical_commit_sink.rs

@@ -179,6 +179,7 @@ impl IPhysicalPlan for CommitSink {
                                         block_meta: Arc::unwrap_or_clone(block_meta),
                                         draft_virtual_block_meta: None,
                                         column_hlls: column_hlls.map(BlockHLLState::Serialized),
+                                        column_top_n: None,
                                     })
                                 })
                                 .collect::<Vec<Arc<ExtendedBlockMeta>>>();

src/query/service/src/physical_plans/physical_mutation_source.rs

@@ -145,6 +145,7 @@ impl IPhysicalPlan for MutationSource {
                         virtual_schema: None,
                         virtual_schema_mode: VirtualSchemaMode::Merge,
                         hll: HashMap::new(),
+                        top_n: HashMap::new(),
                     };
                     let block = DataBlock::empty_with_meta(Box::new(meta));
                     OneBlockSource::create(output, block)

src/query/service/src/test_kits/fuse.rs

@@ -220,7 +220,7 @@ async fn generate_blocks(
         block_metas.push(Arc::new(block_meta));
         hlls.push(hll);
     }
-    let stats = SegmentStatistics::new(hlls).to_bytes()?;
+    let stats = SegmentStatistics::new(hlls, Vec::new()).to_bytes()?;
     Ok((block_metas, stats))
 }
 

src/query/sql/src/planner/plans/scan.rs

@@ -564,6 +564,7 @@ mod tests {
     fn test_sampled_scan_clears_top_n_stats() -> Result<()> {
         let column = Symbol::new(1);
         let top_n = ColumnTopN {
+            capacity: 1,
             values: vec![ColumnTopNEntry {
                 scalar: Scalar::Number(NumberScalar::UInt64(42)),
                 count: 37,

src/query/sql/tests/it/optimizer/collect_statistics.rs

@@ -41,6 +41,7 @@ async fn test_collect_statistics_skips_top_n_for_change_scan() -> Result<()> {
     ctx.configure_for_optimizer_case(true)?;
 
     let top_n = ColumnTopN {
+        capacity: 1,
         values: vec![ColumnTopNEntry {
             scalar: Scalar::Number(NumberScalar::UInt64(1)),
             count: 90,