feat(query): add clustering depth percentiles and optimize overlap an… (#20023)

* feat(query): add clustering depth percentiles and optimize overlap analysis

* fix

* fix

Author: zhyass

src/query/expression/src/kernels/sort.rs

@@ -110,15 +110,19 @@ impl DataBlock {
     }
 }
 
-pub fn compare_scalars(rows: Vec<Vec<Scalar>>, data_types: &[DataType]) -> Result<Vec<u32>> {
+/// Sort scalar rows without taking ownership of the row values.
+pub fn compare_scalars<T: AsRef<[Scalar]>>(
+    rows: &[T],
+    data_types: &[DataType],
+) -> Result<Vec<u32>> {
     let length = rows.len();
     let mut columns = data_types
         .iter()
         .map(|ty| ColumnBuilder::with_capacity(ty, length))
         .collect::<Vec<_>>();
 
-    for row in rows.into_iter() {
-        for (field, column) in row.into_iter().zip(columns.iter_mut()) {
+    for row in rows {
+        for (field, column) in row.as_ref().iter().zip(columns.iter_mut()) {
             column.push(field.as_ref());
         }
     }

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

@@ -23,8 +23,6 @@ use databend_common_expression::ColumnRef;
 use databend_common_expression::DataBlock;
 use databend_common_expression::Expr;
 use databend_common_expression::Scalar;
-use databend_common_expression::TableSchema;
-use databend_common_expression::TableSchemaRef;
 use databend_common_expression::types::DataType;
 use databend_common_expression::types::NumberDataType;
 use databend_common_storages_fuse::FuseBlockPartInfo;
@@ -285,7 +283,7 @@ async fn target_select_segment_locations_with_mode(
     max_segments: usize,
     mode: ReclusterMode,
 ) -> anyhow::Result<(usize, u64, ReclusterParts)> {
-    let schema = TableSchemaRef::new(TableSchema::empty());
+    let schema = TestFixture::default_table_schema();
     let segment_locations = create_segment_location_vector(segment_locations, None);
     let compact_segments = FuseTable::segment_pruning(
         &ctx,
@@ -457,7 +455,7 @@ async fn test_recluster_mutator_selects_multiple_segment_windows() -> anyhow::Re
     )
     .await?;
 
-    let schema = TableSchemaRef::new(TableSchema::empty());
+    let schema = TestFixture::default_table_schema();
     let ctx: Arc<dyn TableContext> = ctx.clone();
     let segment_locations = create_segment_location_vector(segment_locations, None);
     let compact_segments = FuseTable::segment_pruning(

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

@@ -70,6 +70,7 @@ use crate::operations::mutation::BlockIndex;
 use crate::operations::mutation::SegmentIndex;
 use crate::statistics::VirtualColumnAccumulator;
 use crate::statistics::get_min_max_stats;
+use crate::statistics::prepare_cluster_key_exprs;
 use crate::statistics::reducers::merge_statistics_mut;
 use crate::statistics::reducers::reduce_block_metas;
 use crate::statistics::sort_by_cluster_stats;
@@ -925,12 +926,12 @@ fn fill_missing_segment_cluster_stats(
         return;
     }
 
+    let prepared_cluster_key_exprs = prepare_cluster_key_exprs(cluster_key_exprs, schema);
     let (min, max) = get_min_max_stats(
-        cluster_key_exprs,
+        &prepared_cluster_key_exprs,
         &summary.col_stats,
         None,
         Some(cluster_key_id),
-        schema,
     );
     summary.cluster_stats = Some(ClusterStatistics::new(cluster_key_id, min, max, 0, None));
 }

src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs

@@ -64,7 +64,9 @@ use crate::SegmentLocation;
 use crate::io::MetaReaders;
 use crate::operations::ReclusterMode;
 use crate::operations::common::BlockMetaIndex as BlockIndex;
+use crate::statistics::PreparedClusterKeyExpr;
 use crate::statistics::get_min_max_stats;
+use crate::statistics::prepare_cluster_key_exprs;
 use crate::statistics::reducers::merge_statistics_mut;
 
 /// Maximum recluster depth allowed when only two blocks remain.
@@ -162,7 +164,7 @@ pub struct ReclusterMutator {
     pub(crate) schema: TableSchemaRef,
     pub(crate) max_tasks: usize,
     pub(crate) memory_threshold: usize,
-    pub(crate) cluster_key_exprs: Vec<Expr<usize>>,
+    pub(crate) prepared_cluster_key_exprs: Vec<PreparedClusterKeyExpr>,
     pub(crate) cluster_key_types: Vec<DataType>,
 }
 
@@ -199,6 +201,8 @@ impl ReclusterMutator {
             .iter()
             .map(|v| v.data_type().clone())
             .collect::<Vec<_>>();
+        let prepared_cluster_key_exprs =
+            prepare_cluster_key_exprs(&cluster_key_exprs, schema.as_ref());
 
         Ok(Self {
             ctx,
@@ -209,7 +213,7 @@ impl ReclusterMutator {
             cluster_key_id,
             max_tasks,
             memory_threshold,
-            cluster_key_exprs,
+            prepared_cluster_key_exprs,
             cluster_key_types,
         })
     }
@@ -234,6 +238,8 @@ impl ReclusterMutator {
             .iter()
             .map(|expr| expr.data_type().clone())
             .collect();
+        let prepared_cluster_key_exprs =
+            prepare_cluster_key_exprs(&cluster_key_exprs, schema.as_ref());
         let memory_threshold = ctx
             .get_settings()
             .get_recluster_block_size()
@@ -248,7 +254,7 @@ impl ReclusterMutator {
             cluster_key_id,
             max_tasks,
             memory_threshold,
-            cluster_key_exprs,
+            prepared_cluster_key_exprs,
             cluster_key_types,
         }
     }
@@ -870,7 +876,7 @@ impl ReclusterMutator {
             // flush, only bounded anchors remain eligible for the next window.
             let mut seen_in_hot_range = IndexSet::new();
             let (keys, values): (Vec<_>, Vec<_>) = segment_points.into_iter().unzip();
-            let sorted_indices = compare_scalars(keys, &self.cluster_key_types)?;
+            let sorted_indices = compare_scalars(&keys, &self.cluster_key_types)?;
 
             for idx in sorted_indices {
                 let start = &values[idx as usize].0;
@@ -1046,11 +1052,10 @@ impl ReclusterMutator {
         }
 
         let (min_stats, max_stats) = get_min_max_stats(
-            &self.cluster_key_exprs,
+            &self.prepared_cluster_key_exprs,
             col_stats,
             cluster_stats,
             Some(self.cluster_key_id),
-            self.schema.as_ref(),
         );
 
         ClusterStatistics::new(self.cluster_key_id, min_stats, max_stats, 0, None)
@@ -1180,7 +1185,7 @@ impl ReclusterMutator {
         let mut point_overlaps: Vec<Vec<usize>> = Vec::with_capacity(points_map.len());
         let mut unfinished_intervals = BTreeMap::new();
         let (keys, values): (Vec<_>, Vec<_>) = points_map.into_iter().unzip();
-        let indices = compare_scalars(keys, &self.cluster_key_types)?;
+        let indices = compare_scalars(&keys, &self.cluster_key_types)?;
         for (i, idx) in indices.into_iter().enumerate() {
             let start = &values[idx as usize].0;
             let end = &values[idx as usize].1;