refactor(query): simplify append topn stats commit

Author: KKould

src/query/storages/fuse/src/operations/commit.rs

@@ -522,21 +522,13 @@ impl FuseTable {
         insert_hll: &BlockHLL,
         insert_rows: u64,
         insert_top_n: &BlockTopN,
-        reuse_previous_stats: bool,
         refresh_top_n: bool,
     ) -> Result<TableStatsGenerator> {
-        let empty_snapshot = None;
-        let stats_snapshot = if reuse_previous_stats {
-            snapshot
-        } else {
-            &empty_snapshot
-        };
-        // Extract reusable base stats meta (row_count / hll, etc.) from snapshot.
-        // Overwrite commits build statistics from inserted rows only.
-        let summary = stats_snapshot.summary();
+        // Extract previous stats meta (row_count / hll, etc.) from snapshot.
+        let summary = snapshot.summary();
         let prev_stats_meta = summary.additional_stats_meta.as_ref();
-        // Reusable table statistics file location (if any).
-        let mut prev_stats_location = stats_snapshot.table_statistics_location();
+        // Previous statistics file location (if any).
+        let mut prev_stats_location = snapshot.table_statistics_location();
         let top_n_column_ids = if refresh_top_n && insert_rows > 0 {
             self.append_top_n_columns(self.schema())?
                 .map(|(columns, _)| columns.values().map(|field| field.column_id()).collect())
@@ -559,43 +551,23 @@ impl FuseTable {
             ));
         }
 
-        if insert_hll.is_empty() {
-            let empty_hll = HashMap::new();
-            let table_statistics = self
-                .build_append_top_n_statistics(
-                    stats_snapshot,
-                    top_n_column_ids,
-                    &empty_hll,
-                    insert_top_n,
-                    table_row_count,
-                )
-                .await?;
-            if let Some(stats) = &table_statistics {
-                prev_stats_location = Some(self.new_table_statistics_location(stats)?);
-            }
-
-            return Ok(TableStatsGenerator::new(
-                prev_stats_meta.cloned(),
-                prev_stats_location,
-                0,
-                0,
-                HashMap::new(),
-                table_statistics,
-            ));
-        }
-
-        // Initialize a new HLL with inserted rows
         let mut new_hll = insert_hll.clone();
+        let next_stats_meta = if insert_hll.is_empty() {
+            prev_stats_meta.cloned()
+        } else {
+            None
+        };
         // Calculate updated row_count
-        let (hll_row_count, unstats_rows) = match prev_stats_meta {
+        let (hll_row_count, unstats_rows) = match (!insert_hll.is_empty(), prev_stats_meta) {
+            (false, _) => (0, 0),
             // Case 1: Previous stats exist and already contain HLL → merge directly
-            Some(v) if v.hll.is_some() => {
+            (true, Some(v)) if v.hll.is_some() => {
                 let prev_hll = decode_column_hll(v.hll.as_ref().unwrap())?.unwrap();
                 merge_column_hll_mut(&mut new_hll, &prev_hll);
                 (v.row_count + insert_rows, v.unstats_rows)
             }
             // Case 2: Previous meta has no HLL → need to load from stats file
-            _ => {
+            (true, _) => {
                 if let Some(loc) = &prev_stats_location {
                     let ver = TableMetaLocationGenerator::table_statistics_version(loc);
                     let reader = MetaReaders::table_snapshot_statistics_reader(self.get_operator());
@@ -653,7 +625,7 @@ impl FuseTable {
 
         let table_statistics = if has_top_n_update {
             self.build_append_top_n_statistics(
-                stats_snapshot,
+                snapshot,
                 top_n_column_ids,
                 &new_hll,
                 insert_top_n,
@@ -668,7 +640,7 @@ impl FuseTable {
         }
 
         Ok(TableStatsGenerator::new(
-            None,
+            next_stats_meta,
             prev_stats_location,
             hll_row_count,
             unstats_rows,
@@ -701,28 +673,18 @@ impl FuseTable {
         } else {
             None
         };
-        let empty_top_n = HashMap::new();
-        let (mut top_n, append_top_n) = if let Some((previous, prev_stats)) = fresh_prev_stats {
-            let prev_top_n = (previous.summary.row_count != 0).then_some(PreviousTopN {
-                top_n: &prev_stats.top_n,
-                summary: &previous.summary,
-            });
-            build_append_top_n_merge_inputs(top_n_column_ids, prev_top_n, insert_top_n)
-        } else if let Some(previous) = snapshot.as_ref()
-            && previous.summary.row_count != 0
-        {
-            build_append_top_n_merge_inputs(
-                top_n_column_ids,
-                Some(PreviousTopN {
-                    top_n: &empty_top_n,
-                    summary: &previous.summary,
-                }),
-                insert_top_n,
-            )
-        } else {
-            build_append_top_n_merge_inputs(top_n_column_ids, None, insert_top_n)
-        };
 
+        let mut top_n = fresh_prev_stats
+            .map(|(_, stats)| stats.top_n.clone())
+            .unwrap_or_default();
+        top_n.retain(|column_id, _| top_n_column_ids.contains(column_id));
+
+        let append_top_n = append_top_n_merge_input(
+            &top_n_column_ids,
+            snapshot.as_ref().map(|snapshot| &snapshot.summary),
+            &top_n,
+            insert_top_n,
+        );
         merge_column_top_n_mut(&mut top_n, append_top_n)?;
         if top_n.is_empty() {
             return Ok(None);
@@ -753,58 +715,40 @@ impl FuseTable {
     }
 }
 
-#[derive(Clone, Copy)]
-struct PreviousTopN<'a> {
-    top_n: &'a BlockTopN,
-    summary: &'a Statistics,
-}
-
-impl PreviousTopN<'_> {
-    fn missing_error(&self, column_id: ColumnId) -> Option<u64> {
-        self.summary
-            .col_stats
-            .get(&column_id)
-            .map(|stats| self.summary.row_count.saturating_sub(stats.null_count))
-    }
-}
-
-fn build_append_top_n_merge_inputs(
-    top_n_column_ids: Vec<ColumnId>,
-    prev_top_n: Option<PreviousTopN<'_>>,
+fn append_top_n_merge_input(
+    column_ids: &[ColumnId],
+    previous_summary: Option<&Statistics>,
+    previous_top_n: &BlockTopN,
     insert_top_n: &BlockTopN,
-) -> (BlockTopN, BlockTopN) {
-    let mut top_n = HashMap::new();
+) -> BlockTopN {
+    let previous_summary = previous_summary.filter(|summary| summary.row_count != 0);
     let mut append_top_n = HashMap::new();
 
-    for column_id in top_n_column_ids {
-        match prev_top_n {
-            Some(prev_top_n) => {
-                let append_column_top_n = insert_top_n.get(&column_id).cloned();
-                if let Some(column_top_n) = prev_top_n.top_n.get(&column_id).cloned() {
-                    top_n.insert(column_id, column_top_n);
-                    if let Some(column_top_n) = append_column_top_n {
-                        append_top_n.insert(column_id, column_top_n);
-                    }
-                } else if let Some(mut column_top_n) = append_column_top_n {
-                    let Some(missing_error) = prev_top_n.missing_error(column_id) else {
-                        continue;
-                    };
-                    for entry in &mut column_top_n.values {
-                        entry.count = entry.count.saturating_add(missing_error);
-                        entry.error = entry.error.saturating_add(missing_error);
-                    }
-                    append_top_n.insert(column_id, column_top_n);
-                }
-            }
-            None => {
-                if let Some(column_top_n) = insert_top_n.get(&column_id).cloned() {
-                    append_top_n.insert(column_id, column_top_n);
-                }
-            }
+    for &column_id in column_ids {
+        let Some(column_top_n) = insert_top_n.get(&column_id).cloned() else {
+            continue;
+        };
+        if previous_top_n.contains_key(&column_id)
+            || column_has_no_previous_values(previous_summary, column_id)
+        {
+            append_top_n.insert(column_id, column_top_n);
         }
     }
 
-    (top_n, append_top_n)
+    append_top_n
+}
+
+fn column_has_no_previous_values(
+    previous_summary: Option<&Statistics>,
+    column_id: ColumnId,
+) -> bool {
+    let Some(summary) = previous_summary else {
+        return true;
+    };
+    summary
+        .col_stats
+        .get(&column_id)
+        .is_none_or(|stats| stats.null_count == summary.row_count)
 }
 
 pub(crate) fn is_fresh_table_snapshot_top_n(

src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs

@@ -411,17 +411,15 @@ async fn write_new_snapshot_and_build_table_meta(
 ) -> Result<(TableMeta, u64)> {
     let fuse_table = FuseTable::try_from_table(table)?;
     let previous = fuse_table.read_table_snapshot().await?;
-    let reuse_previous_stats = !snapshot_generator.is_overwrite();
     // Match single-table commits: transaction commits may collapse intermediate snapshot
     // lineage, so skip append TopN refresh until transaction stats invalidation is defined.
-    let refresh_top_n = !txn_mgr.lock().is_active();
+    let refresh_top_n = !snapshot_generator.is_overwrite() && !txn_mgr.lock().is_active();
     let mut table_stats_gen = fuse_table
         .generate_table_stats(
             &previous,
             insert_hll,
             insert_rows,
             insert_top_n,
-            reuse_previous_stats,
             refresh_top_n,
         )
         .await?;

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -416,15 +416,11 @@ where F: SnapshotGenerator + Send + Sync + 'static
     fn has_insert_top_n_input(&self) -> bool {
         // Transaction commits may collapse intermediate snapshot lineage, so keep append TopN
         // refresh to autocommit inserts until transaction stats invalidation is defined.
-        self.is_append_only_txn() && !self.ctx.txn_mgr().lock().is_active()
-    }
-
-    fn reuse_previous_table_stats(&self) -> bool {
-        !self
-            .snapshot_gen
+        self.snapshot_gen
             .as_any()
             .downcast_ref::<AppendGenerator>()
-            .is_some_and(|g| g.is_overwrite())
+            .is_some_and(|g| !g.is_overwrite())
+            && !self.ctx.txn_mgr().lock().is_active()
     }
 
     /// Append-only inserts (e.g. `INSERT INTO t SELECT ...`) may skip committing if nothing was
@@ -652,7 +648,6 @@ where F: SnapshotGenerator + Send + Sync + 'static
                             &self.insert_hll,
                             self.insert_rows,
                             &self.insert_top_n,
-                            self.reuse_previous_table_stats(),
                             self.has_insert_top_n_input(),
                         )
                         .await?;
@@ -844,7 +839,6 @@ where F: SnapshotGenerator + Send + Sync + 'static
                         &self.insert_hll,
                         self.insert_rows,
                         &self.insert_top_n,
-                        self.reuse_previous_table_stats(),
                         self.has_insert_top_n_input(),
                     )
                     .await?;

tests/sqllogictests/suites/base/09_fuse_engine/09_0053_analyze_top_n.test

@@ -131,7 +131,7 @@ EXPLAIN SELECT * FROM t_edge WHERE a = 1
 ----
 table: default.db_09_0053.t_edge<slt:ignore>estimated rows: 90.00
 
-# Previous rows with no TopN entries but valid column stats can still seed append TopN.
+# Previous rows with no TopN entries but only NULL values can still seed append TopN.
 statement ok
 TRUNCATE TABLE t_edge
 
@@ -153,16 +153,13 @@ EXPLAIN SELECT * FROM t_edge WHERE a = 1
 ----
 table: default.db_09_0053.t_edge<slt:ignore>estimated rows: 90.00
 
-# If old rows lack column stats for a newly configured TopN column, fail closed.
+# If old non-NULL rows have no fresh TopN for a newly configured column, fail closed.
 statement ok
-CREATE TABLE t_edge_default(a int) approx_distinct_columns = ''
+CREATE TABLE t_edge_default(a int, b int) approx_distinct_columns = ''
 
 statement ok
 INSERT INTO t_edge_default
-SELECT number::int FROM numbers(100)
-
-statement ok
-ALTER TABLE t_edge_default ADD COLUMN b int DEFAULT 1
+SELECT number::int, 1::int FROM numbers(100)
 
 statement ok
 ALTER TABLE t_edge_default SET OPTIONS(analyze_top_n_columns = 'b', analyze_top_n_size = 2)
@@ -176,6 +173,29 @@ SELECT snapshot:table_statistics_location::String IS NULL FROM fuse_dump_snapsho
 ----
 1
 
+# Newly added nullable columns have implicit NULL history and can seed append TopN.
+statement ok
+CREATE TABLE t_added_null(a int) approx_distinct_columns = ''
+
+statement ok
+INSERT INTO t_added_null
+SELECT number::int FROM numbers(100)
+
+statement ok
+ALTER TABLE t_added_null ADD COLUMN b int NULL
+
+statement ok
+ALTER TABLE t_added_null SET OPTIONS(analyze_top_n_columns = 'b', analyze_top_n_size = 2)
+
+statement ok
+INSERT INTO t_added_null(a, b)
+SELECT number::int, if(number < 90, 1, number - 88)::int FROM numbers(100)
+
+query T
+EXPLAIN SELECT * FROM t_added_null WHERE b = 1
+----
+table: default.db_09_0053.t_added_null<slt:ignore>estimated rows: 90.00
+
 # Active transactions skip append TopN refresh until transaction stats lifecycle is handled.
 statement ok
 CREATE TABLE t_txn(a int) approx_distinct_columns = '' analyze_top_n_columns = 'a' analyze_top_n_size = 2
@@ -285,7 +305,7 @@ table: default.db_09_0053.t_no_snapshot_stats<slt:ignore>estimated rows: 90.00
 statement ok
 set enable_table_snapshot_stats = 1
 
-# Reuse one analyzed table for append-after-analyze, new hot values, overwrite, and mutation invalidation.
+# Reuse one analyzed table for append-after-analyze, new hot values, and mutation invalidation.
 statement ok
 CREATE TABLE t_analyzed(id int, a int) approx_distinct_columns = '' analyze_top_n_columns = 'a' analyze_top_n_size = 2
 
@@ -342,17 +362,6 @@ FROM (SELECT snapshot FROM fuse_dump_snapshots('db_09_0053', 't_analyzed') LIMIT
 ----
 1
 
-# INSERT OVERWRITE must build stats from overwritten rows only.
-statement ok
-INSERT OVERWRITE t_analyzed
-SELECT number::int, 2::int FROM numbers(100)
-
-query B
-SELECT count(DISTINCT snapshot:table_statistics_location::String) = 2
-FROM (SELECT snapshot FROM fuse_dump_snapshots('db_09_0053', 't_analyzed') LIMIT 2)
-----
-1
-
 # Non-insert mutations should not keep old TopN fresh.
 statement ok
 TRUNCATE TABLE t_analyzed