chore(planner): improve histogram cardinality estimation

src/common/storage/src/statistics.rs

@@ -59,6 +59,14 @@ impl Datum {
         matches!(self, Datum::Bytes(_))
     }
 
+    pub fn to_float(self) -> Self {
+        match self {
+            Datum::Int(v) => Datum::Float(F64::from(v as f64)),
+            Datum::UInt(v) => Datum::Float(F64::from(v as f64)),
+            _ => self,
+        }
+    }
+
     pub fn to_double(&self) -> Result<f64> {
         match self {
             Datum::Bool(v) => Ok(*v as u8 as f64),

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

@@ -147,6 +147,7 @@ impl Interpreter for ExplainInterpreter {
                         self.ctx.clone(),
                         *s_expr.clone(),
                         schema.clone(),
+                        metadata.clone(),
                     )?;
                     let plan = interpreter.build_physical_plan(&mutation, None).await?;
                     self.explain_physical_plan(&plan, metadata, &None).await?
@@ -529,7 +530,12 @@ impl ExplainInterpreter {
         schema: DataSchemaRef,
     ) -> Result<Vec<DataBlock>> {
         let mutation: Mutation = s_expr.plan().clone().try_into()?;
-        let interpreter = MutationInterpreter::try_create(self.ctx.clone(), s_expr, schema)?;
+        let interpreter = MutationInterpreter::try_create(
+            self.ctx.clone(),
+            s_expr,
+            schema,
+            mutation.metadata.clone(),
+        )?;
         let plan = interpreter.build_physical_plan(&mutation, None).await?;
         let root_fragment = Fragmenter::try_create(self.ctx.clone())?.build_fragment(&plan)?;
 

src/query/service/src/interpreters/interpreter_factory.rs

@@ -20,6 +20,7 @@ use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_sql::binder::ExplainConfig;
+use databend_common_sql::plans::Mutation;
 use log::error;
 
 use super::interpreter_catalog_create::CreateCatalogInterpreter;
@@ -467,9 +468,15 @@ impl InterpreterFactory {
             Plan::Insert(insert) => InsertInterpreter::try_create(ctx, *insert.clone()),
 
             Plan::Replace(replace) => ReplaceInterpreter::try_create(ctx, *replace.clone()),
-            Plan::DataMutation { s_expr, schema, .. } => Ok(Arc::new(
-                MutationInterpreter::try_create(ctx, *s_expr.clone(), schema.clone())?,
-            )),
+            Plan::DataMutation { s_expr, schema, .. } => {
+                let mutation: Mutation = s_expr.plan().clone().try_into()?;
+                Ok(Arc::new(MutationInterpreter::try_create(
+                    ctx,
+                    *s_expr.clone(),
+                    schema.clone(),
+                    mutation.metadata.clone(),
+                )?))
+            }
 
             // Roles
             Plan::CreateRole(create_role) => Ok(Arc::new(CreateRoleInterpreter::try_create(

src/query/service/src/interpreters/interpreter_mutation.rs

@@ -36,6 +36,7 @@ use databend_common_sql::executor::MutationBuildInfo;
 use databend_common_sql::executor::PhysicalPlan;
 use databend_common_sql::executor::PhysicalPlanBuilder;
 use databend_common_sql::optimizer::SExpr;
+use databend_common_sql::planner::MetadataRef;
 use databend_common_sql::plans;
 use databend_common_sql::plans::Mutation;
 use databend_common_storage::MutationStatus;
@@ -44,6 +45,7 @@ use databend_common_storages_fuse::operations::TruncateMode;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
 use databend_storages_common_table_meta::meta::TableSnapshot;
+use log::info;
 
 use crate::interpreters::common::check_deduplicate_label;
 use crate::interpreters::common::dml_build_update_stream_req;
@@ -58,18 +60,21 @@ pub struct MutationInterpreter {
     ctx: Arc<QueryContext>,
     s_expr: SExpr,
     schema: DataSchemaRef,
+    metadata: MetadataRef,
 }
 
 impl MutationInterpreter {
     pub fn try_create(
         ctx: Arc<QueryContext>,
         s_expr: SExpr,
         schema: DataSchemaRef,
+        metadata: MetadataRef,
     ) -> Result<MutationInterpreter> {
         Ok(MutationInterpreter {
             ctx,
             s_expr,
             schema,
+            metadata,
         })
     }
 }
@@ -129,6 +134,12 @@ impl Interpreter for MutationInterpreter {
             .build_physical_plan(&mutation, Some(mutation_build_info))
             .await?;
 
+        let query_plan = physical_plan
+            .format(self.metadata.clone(), Default::default())?
+            .format_pretty()?;
+
+        info!("Query physical plan: \n{}", query_plan);
+
         // Build pipeline.
         let mut build_res =
             build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;

src/query/service/src/interpreters/interpreter_table_add_column.rs

@@ -27,6 +27,7 @@ use databend_common_meta_types::MatchSeq;
 use databend_common_sql::field_default_value;
 use databend_common_sql::plans::AddColumnOption;
 use databend_common_sql::plans::AddTableColumnPlan;
+use databend_common_sql::plans::Mutation;
 use databend_common_sql::plans::Plan;
 use databend_common_sql::Planner;
 use databend_common_storages_fuse::FuseTable;
@@ -135,8 +136,13 @@ impl Interpreter for AddTableColumnInterpreter {
             let mut planner = Planner::new(self.ctx.clone());
             let (plan, _) = planner.plan_sql(&query).await?;
             if let Plan::DataMutation { s_expr, schema, .. } = plan {
-                let interpreter =
-                    MutationInterpreter::try_create(self.ctx.clone(), *s_expr, schema)?;
+                let mutation: Mutation = s_expr.plan().clone().try_into()?;
+                let interpreter = MutationInterpreter::try_create(
+                    self.ctx.clone(),
+                    *s_expr,
+                    schema,
+                    mutation.metadata.clone(),
+                )?;
                 let _ = interpreter.execute(self.ctx.clone()).await?;
                 return Ok(PipelineBuildResult::create());
             }

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

@@ -27,6 +27,7 @@ use databend_common_expression::DataSchemaRef;
 use databend_common_expression::ScalarRef;
 use databend_common_expression::SendableDataBlockStream;
 use databend_common_sql::optimizer::SExpr;
+use databend_common_sql::plans::Mutation;
 use databend_common_storages_factory::Table;
 use databend_common_storages_fuse::io::MetaWriter;
 use databend_common_storages_fuse::statistics::gen_columns_statistics;
@@ -286,7 +287,9 @@ pub async fn do_mutation(
     s_expr: SExpr,
     schema: DataSchemaRef,
 ) -> Result<()> {
-    let interpreter = MutationInterpreter::try_create(ctx.clone(), s_expr, schema)?;
+    let mutation: Mutation = s_expr.plan().clone().try_into()?;
+    let interpreter =
+        MutationInterpreter::try_create(ctx.clone(), s_expr, schema, mutation.metadata.clone())?;
     let _ = interpreter.execute(ctx).await?;
     Ok(())
 }

src/query/sql/src/planner/optimizer/property/histogram.rs

@@ -14,12 +14,15 @@
 
 use std::cmp::Ordering;
 
+use databend_common_base::base::OrderedFloat;
 use databend_common_exception::Result;
 use databend_common_expression::arithmetics_type::ResultTypeOfUnary;
 use databend_common_storage::Datum;
 use databend_common_storage::Histogram;
 use databend_common_storage::HistogramBucket;
 
+pub type F64 = OrderedFloat<f64>;
+
 /// Construct a histogram from NDV and total number of rows.
 ///
 /// # Arguments
@@ -59,7 +62,7 @@ pub fn histogram_from_ndv(
     }
 
     let (min, max) = match bound {
-        Some((min, max)) => (min, max),
+        Some((min, max)) => (min.to_float(), max.to_float()),
         None => {
             return Err(format!(
                 "Must have min and max value when NDV is greater than 0, got NDV: {}",
@@ -182,7 +185,7 @@ impl SampleSet for UniformSampleSet {
 
             (Datum::Float(min), Datum::Float(max)) => {
                 let min = *min;
-                let max = *max;
+                let max = (*max).checked_add(F64::from(1.0)).ok_or("overflowed")?;
                 // TODO(xudong): better histogram computation.
                 let bucket_range = max.checked_sub(min).ok_or("overflowed")? / num_buckets as f64;
                 let upper_bound = min + bucket_range * bucket_index as f64;

tests/sqllogictests/suites/mode/cluster/memo/aggregate_property.test

@@ -63,22 +63,22 @@ Memo
 │   └── #0 Join [#0, #3]
 ├── Group #5
 │   ├── Best properties
-│   │   ├── { dist: Any }: expr: #0, cost: 4419.000, children: [{ dist: Any }]
-│   │   └── { dist: Serial }: expr: #1, cost: 7569.000, children: [{ dist: Any }]
+│   │   ├── { dist: Any }: expr: #0, cost: 4420.000, children: [{ dist: Any }]
+│   │   └── { dist: Serial }: expr: #1, cost: 7920.000, children: [{ dist: Any }]
 │   ├── #0 EvalScalar [#4]
 │   └── #1 Exchange: (Merge) [#5]
 ├── Group #6
 │   ├── Best properties
-│   │   └── { dist: Serial }: expr: #0, cost: 7614.000, children: [{ dist: Serial }]
+│   │   └── { dist: Serial }: expr: #0, cost: 7970.000, children: [{ dist: Serial }]
 │   ├── #0 Aggregate [#5]
 │   └── #1 Exchange: (Merge) [#6]
 ├── Group #7
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 7619.000, children: [{ dist: Serial }]
+│   │   └── { dist: Any }: expr: #0, cost: 7975.000, children: [{ dist: Serial }]
 │   └── #0 Aggregate [#6]
 └── Group #8
     ├── Best properties
-    │   └── { dist: Serial }: expr: #0, cost: 7620.000, children: [{ dist: Any }]
+    │   └── { dist: Serial }: expr: #0, cost: 7976.000, children: [{ dist: Any }]
     └── #0 EvalScalar [#7]
 
 query T
@@ -126,22 +126,22 @@ Memo
 │   └── #0 Join [#0, #3]
 ├── Group #5
 │   ├── Best properties
-│   │   ├── { dist: Any }: expr: #0, cost: 4419.000, children: [{ dist: Any }]
-│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 4878.000, children: [{ dist: Any }]
+│   │   ├── { dist: Any }: expr: #0, cost: 4420.000, children: [{ dist: Any }]
+│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 4930.000, children: [{ dist: Any }]
 │   ├── #0 EvalScalar [#4]
 │   └── #1 Exchange: (Hash(t_10.a (#0)::Int32 NULL)) [#5]
 ├── Group #6
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 4923.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
+│   │   └── { dist: Any }: expr: #0, cost: 4980.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
 │   └── #0 Aggregate [#5]
 ├── Group #7
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 4968.000, children: [{ dist: Any }]
+│   │   └── { dist: Any }: expr: #0, cost: 5030.000, children: [{ dist: Any }]
 │   └── #0 Aggregate [#6]
 ├── Group #8
 │   ├── Best properties
-│   │   ├── { dist: Any }: expr: #0, cost: 4977.000, children: [{ dist: Any }]
-│   │   └── { dist: Serial }: expr: #4, cost: 8127.000, children: [{ dist: Any }]
+│   │   ├── { dist: Any }: expr: #0, cost: 5040.000, children: [{ dist: Any }]
+│   │   └── { dist: Serial }: expr: #4, cost: 8540.000, children: [{ dist: Any }]
 │   ├── #0 EvalScalar [#7]
 │   ├── #1 EvalScalar [#14]
 │   ├── #2 EvalScalar [#20]
@@ -166,16 +166,16 @@ Memo
 ├── Group #12
 │   ├── Best properties
 │   │   ├── { dist: Any }: expr: #0, cost: 66410.000, children: [{ dist: Any }, { dist: Broadcast }]
-│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 66869.000, children: [{ dist: Any }]
+│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 66920.000, children: [{ dist: Any }]
 │   ├── #0 Join [#11, #3]
 │   └── #1 Exchange: (Hash(t_10.a (#0)::Int32 NULL)) [#12]
 ├── Group #13
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 66914.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
+│   │   └── { dist: Any }: expr: #0, cost: 66970.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
 │   └── #0 Aggregate [#12]
 ├── Group #14
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 66959.000, children: [{ dist: Any }]
+│   │   └── { dist: Any }: expr: #0, cost: 67020.000, children: [{ dist: Any }]
 │   └── #0 Aggregate [#13]
 ├── Group #15
 │   ├── Best properties
@@ -197,35 +197,35 @@ Memo
 │   └── #0 Join [#0, #16]
 ├── Group #18
 │   ├── Best properties
-│   │   ├── { dist: Any }: expr: #0, cost: 5029.000, children: [{ dist: Any }]
-│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 5488.000, children: [{ dist: Any }]
+│   │   ├── { dist: Any }: expr: #0, cost: 5030.000, children: [{ dist: Any }]
+│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 5540.000, children: [{ dist: Any }]
 │   ├── #0 EvalScalar [#17]
 │   └── #1 Exchange: (Hash(t_10.a (#0)::Int32 NULL)) [#18]
 ├── Group #19
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 5533.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
+│   │   └── { dist: Any }: expr: #0, cost: 5590.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
 │   └── #0 Aggregate [#18]
 ├── Group #20
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 5578.000, children: [{ dist: Any }]
+│   │   └── { dist: Any }: expr: #0, cost: 5640.000, children: [{ dist: Any }]
 │   └── #0 Aggregate [#19]
 ├── Group #21
 │   ├── Best properties
 │   │   └── { dist: Any }: expr: #0, cost: 67020.000, children: [{ dist: Any }, { dist: Broadcast }]
 │   └── #0 Join [#11, #16]
 ├── Group #22
 │   ├── Best properties
-│   │   ├── { dist: Any }: expr: #0, cost: 67029.000, children: [{ dist: Any }]
-│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 67488.000, children: [{ dist: Any }]
+│   │   ├── { dist: Any }: expr: #0, cost: 67030.000, children: [{ dist: Any }]
+│   │   └── { dist: Hash(t_10.a (#0)::Int32 NULL) }: expr: #1, cost: 67540.000, children: [{ dist: Any }]
 │   ├── #0 EvalScalar [#21]
 │   └── #1 Exchange: (Hash(t_10.a (#0)::Int32 NULL)) [#22]
 ├── Group #23
 │   ├── Best properties
-│   │   └── { dist: Any }: expr: #0, cost: 67533.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
+│   │   └── { dist: Any }: expr: #0, cost: 67590.000, children: [{ dist: Hash(t_10.a (#0)::Int32 NULL) }]
 │   └── #0 Aggregate [#22]
 └── Group #24
     ├── Best properties
-    │   └── { dist: Any }: expr: #0, cost: 67578.000, children: [{ dist: Any }]
+    │   └── { dist: Any }: expr: #0, cost: 67640.000, children: [{ dist: Any }]
     └── #0 Aggregate [#23]
 
 

tests/sqllogictests/suites/mode/cluster/memo/join_property.test

@@ -62,8 +62,8 @@ Memo
 │   └── #0 Join [#0, #3]
 └── Group #5
     ├── Best properties
-    │   ├── { dist: Any }: expr: #0, cost: 4419.000, children: [{ dist: Any }]
-    │   └── { dist: Serial }: expr: #1, cost: 7569.000, children: [{ dist: Any }]
+    │   ├── { dist: Any }: expr: #0, cost: 4420.000, children: [{ dist: Any }]
+    │   └── { dist: Serial }: expr: #1, cost: 7920.000, children: [{ dist: Any }]
     ├── #0 EvalScalar [#4]
     └── #1 Exchange: (Merge) [#5]
 
@@ -192,8 +192,8 @@ Memo
 │   └── #0 Join [#2, #3]
 └── Group #5
     ├── Best properties
-    │   ├── { dist: Any }: expr: #0, cost: 112911.000, children: [{ dist: Any }]
-    │   └── { dist: Serial }: expr: #1, cost: 494761.000, children: [{ dist: Any }]
+    │   ├── { dist: Any }: expr: #0, cost: 112910.000, children: [{ dist: Any }]
+    │   └── { dist: Serial }: expr: #1, cost: 494410.000, children: [{ dist: Any }]
     ├── #0 EvalScalar [#4]
     └── #1 Exchange: (Merge) [#5]