feat(storage): analyze table noscan (#18254)

* analyze table noscan

* fix

* add test

* fix test

* fix test

* fix test

* remove unused codes

Author: zhyass

src/query/ast/src/ast/statements/table.rs

@@ -724,6 +724,7 @@ pub struct AnalyzeTableStmt {
     pub catalog: Option<Identifier>,
     pub database: Option<Identifier>,
     pub table: Identifier,
+    pub no_scan: bool,
 }
 
 impl Display for AnalyzeTableStmt {
@@ -736,6 +737,9 @@ impl Display for AnalyzeTableStmt {
                 .chain(&self.database)
                 .chain(Some(&self.table)),
         )?;
+        if self.no_scan {
+            write!(f, " NOSCAN")?;
+        }
 
         Ok(())
     }

src/query/ast/src/parser/statement.rs

@@ -1170,13 +1170,14 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
     );
     let analyze_table = map(
         rule! {
-            ANALYZE ~ TABLE ~ #dot_separated_idents_1_to_3
+            ANALYZE ~ TABLE ~ #dot_separated_idents_1_to_3 ~ NOSCAN?
         },
-        |(_, _, (catalog, database, table))| {
+        |(_, _, (catalog, database, table), no_scan)| {
             Statement::AnalyzeTable(AnalyzeTableStmt {
                 catalog,
                 database,
                 table,
+                no_scan: no_scan.is_some(),
             })
         },
     );

src/query/ast/src/parser/token.rs

@@ -868,6 +868,8 @@ pub enum TokenKind {
     NO_PASSWORD,
     #[token("NONE", ignore(ascii_case))]
     NONE,
+    #[token("NOSCAN", ignore(ascii_case))]
+    NOSCAN,
     #[token("NOT", ignore(ascii_case))]
     NOT,
     #[token("NOTENANTSETTING", ignore(ascii_case))]

src/query/service/src/interpreters/interpreter_table_analyze.rs

@@ -35,6 +35,7 @@ use databend_common_sql::Planner;
 use databend_common_storage::DEFAULT_HISTOGRAM_BUCKETS;
 use databend_common_storages_factory::NavigationPoint;
 use databend_common_storages_factory::Table;
+use databend_common_storages_fuse::operations::AnalyzeLightMutator;
 use databend_common_storages_fuse::operations::HistogramInfoSink;
 use databend_common_storages_fuse::FuseTable;
 use databend_storages_common_index::Index;
@@ -109,66 +110,77 @@ impl Interpreter for AnalyzeTableInterpreter {
             Err(_) => return Ok(PipelineBuildResult::create()),
         };
 
-        let r = table.read_table_snapshot().await;
-        let snapshot_opt = match r {
-            Err(e) => return Err(e),
-            Ok(v) => v,
+        let Some(snapshot) = table.read_table_snapshot().await? else {
+            return Ok(PipelineBuildResult::create());
         };
 
-        if let Some(snapshot) = snapshot_opt {
-            // plan sql
-            let _table_info = table.get_table_info();
+        if self.plan.no_scan {
+            let operator = table.get_operator();
+            let cluster_key_id = table.cluster_key_id();
+            let table_meta_timestamps = self
+                .ctx
+                .get_table_meta_timestamps(table, Some(snapshot.clone()))?;
+            let mut mutator = AnalyzeLightMutator::create(
+                self.ctx.clone(),
+                operator,
+                snapshot,
+                cluster_key_id,
+                table_meta_timestamps,
+            );
+            mutator.target_select().await?;
+            mutator.try_commit(table).await?;
+            return Ok(PipelineBuildResult::create());
+        }
 
-            let table_statistics = table
-                .read_table_snapshot_statistics(Some(&snapshot))
-                .await?;
+        let table_statistics = table
+            .read_table_snapshot_statistics(Some(&snapshot))
+            .await?;
 
-            let (is_full, temporal_str) = if let Some(table_statistics) = &table_statistics {
-                let is_full = match table
-                    .navigate_to_point(
-                        &NavigationPoint::SnapshotID(
-                            table_statistics.snapshot_id.simple().to_string(),
-                        ),
-                        self.ctx.clone().get_abort_checker(),
-                    )
+        // plan sql
+        let (is_full, temporal_str) = if let Some(table_statistics) = &table_statistics {
+            let is_full = match table
+                .navigate_to_point(
+                    &NavigationPoint::SnapshotID(table_statistics.snapshot_id.simple().to_string()),
+                    self.ctx.clone().get_abort_checker(),
+                )
+                .await
+            {
+                Ok(t) => !t
+                    .read_table_snapshot()
                     .await
-                {
-                    Ok(t) => !t
-                        .read_table_snapshot()
-                        .await
-                        .is_ok_and(|s| s.is_some_and(|s| s.prev_table_seq.is_some())),
-                    Err(_) => true,
-                };
+                    .is_ok_and(|s| s.is_some_and(|s| s.prev_table_seq.is_some())),
+                Err(_) => true,
+            };
 
-                let temporal_str = if is_full {
-                    format!("AT (snapshot => '{}')", snapshot.snapshot_id.simple())
-                } else {
-                    // analyze only need to collect the added blocks.
-                    let table_alias = format!("_change_insert${:08x}", Utc::now().timestamp());
-                    format!(
+            let temporal_str = if is_full {
+                format!("AT (snapshot => '{}')", snapshot.snapshot_id.simple())
+            } else {
+                // analyze only need to collect the added blocks.
+                let table_alias = format!("_change_insert${:08x}", Utc::now().timestamp());
+                format!(
                         "CHANGES(INFORMATION => DEFAULT) AT (snapshot => '{}') END (snapshot => '{}') AS {table_alias}",
                         table_statistics.snapshot_id.simple(),
                         snapshot.snapshot_id.simple(),
                     )
-                };
-                (is_full, temporal_str)
-            } else {
-                (
-                    true,
-                    format!("AT (snapshot => '{}')", snapshot.snapshot_id.simple()),
-                )
             };
+            (is_full, temporal_str)
+        } else {
+            (
+                true,
+                format!("AT (snapshot => '{}')", snapshot.snapshot_id.simple()),
+            )
+        };
 
-            let quote = self
-                .ctx
-                .get_settings()
-                .get_sql_dialect()?
-                .default_ident_quote();
+        let quote = self
+            .ctx
+            .get_settings()
+            .get_sql_dialect()?
+            .default_ident_quote();
 
-            // 0.01625 --> 12 buckets --> 4K size per column
-            // 1.04 / math.sqrt(1<<12) --> 0.01625
-            const DISTINCT_ERROR_RATE: f64 = 0.01625;
-            let ndv_select_expr = snapshot
+        // 0.01625 --> 12 buckets --> 4K size per column
+        // 1.04 / math.sqrt(1<<12) --> 0.01625
+        const DISTINCT_ERROR_RATE: f64 = 0.01625;
+        let ndv_select_expr = snapshot
                 .schema
                 .fields()
                 .iter()
@@ -182,22 +194,22 @@ impl Interpreter for AnalyzeTableInterpreter {
                 })
                 .join(", ");
 
-            let sql = format!(
-                "SELECT {ndv_select_expr}, {is_full} as is_full from {}.{} {temporal_str}",
-                plan.database, plan.table,
-            );
+        let sql = format!(
+            "SELECT {ndv_select_expr}, {is_full} as is_full from {}.{} {temporal_str}",
+            plan.database, plan.table,
+        );
 
-            info!("Analyze via sql: {sql}");
+        info!("Analyze via sql: {sql}");
 
-            let (physical_plan, bind_context) = self.plan_sql(sql).await?;
-            let mut build_res =
-                build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;
-            // After profiling, computing histogram is heavy and the bottleneck is window function(90%).
-            // It's possible to OOM if the table is too large and spilling isn't enabled.
-            // We add a setting `enable_analyze_histogram` to control whether to compute histogram(default is closed).
-            let mut histogram_info_receivers = HashMap::new();
-            if self.ctx.get_settings().get_enable_analyze_histogram()? {
-                let histogram_sqls = table
+        let (physical_plan, bind_context) = self.plan_sql(sql).await?;
+        let mut build_res =
+            build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;
+        // After profiling, computing histogram is heavy and the bottleneck is window function(90%).
+        // It's possible to OOM if the table is too large and spilling isn't enabled.
+        // We add a setting `enable_analyze_histogram` to control whether to compute histogram(default is closed).
+        let mut histogram_info_receivers = HashMap::new();
+        if self.ctx.get_settings().get_enable_analyze_histogram()? {
+            let histogram_sqls = table
                     .schema()
                     .fields()
                     .iter()
@@ -222,50 +234,47 @@ impl Interpreter for AnalyzeTableInterpreter {
                         )
                     })
                     .collect::<Vec<_>>();
-                for (sql, col_id) in histogram_sqls.into_iter() {
-                    info!("Analyze histogram via sql: {sql}");
-                    let (mut histogram_plan, bind_context) = self.plan_sql(sql).await?;
-                    if !self.ctx.get_cluster().is_empty() {
-                        histogram_plan = remove_exchange(histogram_plan);
-                    }
-                    let mut histogram_build_res = build_query_pipeline(
-                        &QueryContext::create_from(self.ctx.as_ref()),
-                        &bind_context.columns,
-                        &histogram_plan,
-                        false,
-                    )
-                    .await?;
-                    let (tx, rx) = async_channel::unbounded();
-                    histogram_build_res.main_pipeline.add_sink(|input_port| {
-                        Ok(ProcessorPtr::create(HistogramInfoSink::create(
-                            Some(tx.clone()),
-                            input_port.clone(),
-                        )))
-                    })?;
-
-                    build_res
-                        .sources_pipelines
-                        .push(histogram_build_res.main_pipeline.finalize(None));
-                    build_res
-                        .sources_pipelines
-                        .extend(histogram_build_res.sources_pipelines);
-                    histogram_info_receivers.insert(col_id, rx);
+            for (sql, col_id) in histogram_sqls.into_iter() {
+                info!("Analyze histogram via sql: {sql}");
+                let (mut histogram_plan, bind_context) = self.plan_sql(sql).await?;
+                if !self.ctx.get_cluster().is_empty() {
+                    histogram_plan = remove_exchange(histogram_plan);
                 }
+                let mut histogram_build_res = build_query_pipeline(
+                    &QueryContext::create_from(self.ctx.as_ref()),
+                    &bind_context.columns,
+                    &histogram_plan,
+                    false,
+                )
+                .await?;
+                let (tx, rx) = async_channel::unbounded();
+                histogram_build_res.main_pipeline.add_sink(|input_port| {
+                    Ok(ProcessorPtr::create(HistogramInfoSink::create(
+                        Some(tx.clone()),
+                        input_port.clone(),
+                    )))
+                })?;
+
+                build_res
+                    .sources_pipelines
+                    .push(histogram_build_res.main_pipeline.finalize(None));
+                build_res
+                    .sources_pipelines
+                    .extend(histogram_build_res.sources_pipelines);
+                histogram_info_receivers.insert(col_id, rx);
             }
-            FuseTable::do_analyze(
-                self.ctx.clone(),
-                bind_context.output_schema(),
-                &self.plan.catalog,
-                &self.plan.database,
-                &self.plan.table,
-                snapshot.snapshot_id,
-                &mut build_res.main_pipeline,
-                histogram_info_receivers,
-            )?;
-            return Ok(build_res);
         }
-
-        return Ok(PipelineBuildResult::create());
+        FuseTable::do_analyze(
+            self.ctx.clone(),
+            bind_context.output_schema(),
+            &self.plan.catalog,
+            &self.plan.database,
+            &self.plan.table,
+            snapshot.snapshot_id,
+            &mut build_res.main_pipeline,
+            histogram_info_receivers,
+        )?;
+        Ok(build_res)
     }
 }
 

src/query/service/tests/it/storages/fuse/operations/mutation/segments_compact_mutator.rs

@@ -89,7 +89,7 @@ async fn test_compact_segment_normal_case() -> Result<()> {
     let mutator = build_mutator(fuse_table, ctx.clone(), None).await?;
     assert!(mutator.is_some());
     let mutator = mutator.unwrap();
-    mutator.try_commit(table.clone()).await?;
+    mutator.try_commit(fuse_table).await?;
 
     // check segment count
     let qry = "select segment_count as count from fuse_snapshot('default', 't') limit 1";
@@ -134,7 +134,7 @@ async fn test_compact_segment_resolvable_conflict() -> Result<()> {
     let num_inserts = 9;
     fixture.append_rows(num_inserts).await?;
 
-    mutator.try_commit(table.clone()).await?;
+    mutator.try_commit(fuse_table).await?;
 
     // check segment count
     let count_seg = "select segment_count as count from fuse_snapshot('default', 't') limit 1";
@@ -194,7 +194,7 @@ async fn test_compact_segment_unresolvable_conflict() -> Result<()> {
     }
 
     // the compact operation committed latter should be failed.
-    let r = mutator.try_commit(table.clone()).await;
+    let r = mutator.try_commit(fuse_table).await;
     assert!(r.is_err());
     assert_eq!(r.err().unwrap().code(), ErrorCode::UNRESOLVABLE_CONFLICT);
 
@@ -232,7 +232,7 @@ async fn check_count(result_stream: SendableDataBlockStream) -> Result<u64> {
 pub async fn compact_segment(ctx: Arc<QueryContext>, table: &Arc<dyn Table>) -> Result<()> {
     let fuse_table = FuseTable::try_from_table(table.as_ref())?;
     let mutator = build_mutator(fuse_table, ctx.clone(), None).await?.unwrap();
-    mutator.try_commit(table.clone()).await
+    mutator.try_commit(fuse_table).await
 }
 
 async fn build_mutator(

src/query/sql/src/planner/binder/ddl/table.rs

@@ -1396,6 +1396,7 @@ impl Binder {
             catalog,
             database,
             table,
+            no_scan,
         } = stmt;
 
         let (catalog, database, table) =
@@ -1405,6 +1406,7 @@ impl Binder {
             catalog,
             database,
             table,
+            no_scan: *no_scan,
         })))
     }
 

src/query/sql/src/planner/plans/ddl/table.rs

@@ -205,6 +205,7 @@ pub struct AnalyzeTablePlan {
     pub catalog: String,
     pub database: String,
     pub table: String,
+    pub no_scan: bool,
 }
 
 impl AnalyzeTablePlan {

src/query/storages/common/table_meta/src/meta/column_oriented_segment/segment_builder.rs

@@ -349,9 +349,13 @@ impl ColumnOrientedSegmentBuilder {
                 .unwrap_or(Scalar::Null);
             let null_count = col_stat.null_count.iter().sum();
             let in_memory_size = col_stat.in_memory_size.iter().sum();
+            let distinct_of_values = col_stat
+                .distinct_of_values
+                .iter()
+                .try_fold(0, |acc, ndv| ndv.map(|v| acc + v));
             col_stats.insert(
                 col_id,
-                ColumnStatistics::new(min, max, null_count, in_memory_size, None),
+                ColumnStatistics::new(min, max, null_count, in_memory_size, distinct_of_values),
             );
             col_stat.min = ColumnBuilder::from_column(mins);
             col_stat.max = ColumnBuilder::from_column(maxs);

src/query/storages/fuse/src/fuse_column.rs

@@ -36,12 +36,14 @@ impl FuseTableColumnStatisticsProvider {
         column_distinct_values: Option<HashMap<ColumnId, u64>>,
         row_count: u64,
     ) -> Self {
+        let distinct_map = column_distinct_values.as_ref();
         let column_stats = column_stats
             .into_iter()
             .map(|(column_id, stat)| {
-                let ndv = column_distinct_values.as_ref().map_or(row_count, |map| {
-                    map.get(&column_id).map_or(row_count, |v| *v)
-                });
+                let ndv = distinct_map
+                    .and_then(|map| map.get(&column_id).cloned())
+                    .or(stat.distinct_of_values)
+                    .unwrap_or(row_count);
                 let stat = BasicColumnStatistics {
                     min: Datum::from_scalar(stat.min),
                     max: Datum::from_scalar(stat.max),