fix

Author: zhyass

src/query/service/src/interpreters/interpreter_replace.rs

@@ -47,6 +47,7 @@ use databend_common_storage::StageFileInfo;
 use databend_common_storages_factory::Table;
 use databend_common_storages_fuse::FuseTable;
 use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
+use databend_storages_common_table_meta::table::ClusterType;
 use parking_lot::RwLock;
 
 use crate::interpreters::common::check_deduplicate_label;
@@ -290,7 +291,10 @@ impl ReplaceInterpreter {
  .ctx
  .get_settings()
  .get_replace_into_bloom_pruning_max_column_number()?;
- let bloom_filter_column_indexes = if table.cluster_key_meta().is_some() {
+ let bloom_filter_column_indexes = if table
+ .cluster_type()
+ .is_some_and(|v| v == ClusterType::Linear)
+ {
  fuse_table
  .choose_bloom_filter_columns(
  self.ctx.clone(),

src/query/service/src/pipelines/builders/builder_commit.rs

@@ -40,7 +40,7 @@ impl PipelineBuilder {
  self.main_pipeline.add_async_accumulating_transformer(|| {
  let base_segments = if matches!(
  plan.mutation_kind,
- MutationKind::Compact | MutationKind::Insert | MutationKind::Recluster(_)
+ MutationKind::Compact | MutationKind::Insert | MutationKind::Recluster
  ) {
  vec![]
  } else {

src/query/service/src/pipelines/builders/builder_hilbert_serialize.rs

@@ -21,7 +21,6 @@ use databend_common_storages_fuse::operations::TransformSerializeBlock;
 use databend_common_storages_fuse::statistics::ClusterStatsGenerator;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
-use databend_storages_common_table_meta::table::ClusterType;
 
 use crate::pipelines::PipelineBuilder;
 
@@ -49,7 +48,7 @@ impl PipelineBuilder {
  transform_output_port,
  table,
  ClusterStatsGenerator::default(),
- MutationKind::Recluster(ClusterType::Hilbert),
+ MutationKind::Recluster,
  )?;
  proc.into_processor()
  })

src/query/service/src/pipelines/builders/builder_recluster.rs

@@ -34,7 +34,6 @@ use databend_common_storages_factory::Table;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
-use databend_storages_common_table_meta::table::ClusterType;
 
 use crate::pipelines::builders::SortPipelineBuilder;
 use crate::pipelines::processors::TransformAddStreamColumns;
@@ -178,7 +177,7 @@ impl PipelineBuilder {
  transform_output_port,
  table,
  cluster_stats_gen.clone(),
- MutationKind::Recluster(ClusterType::Linear),
+ MutationKind::Recluster,
  )?;
  proc.into_processor()
  })

src/query/sql/src/executor/physical_plans/common.rs

@@ -17,7 +17,6 @@ use std::fmt::Formatter;
 
 use databend_common_expression::types::DataType;
 use databend_common_expression::Scalar;
-use databend_storages_common_table_meta::table::ClusterType;
 
 use crate::plans::UDFField;
 use crate::plans::UDFType;
@@ -71,7 +70,7 @@ pub enum MutationKind {
  Delete,
  Update,
  Replace,
- Recluster(ClusterType),
+ Recluster,
  Insert,
  Compact,
  MergeInto,
@@ -82,10 +81,7 @@ impl Display for MutationKind {
  match self {
  MutationKind::Delete => write!(f, "Delete"),
  MutationKind::Insert => write!(f, "Insert"),
- MutationKind::Recluster(cluster_type) => match cluster_type {
- ClusterType::Linear => write!(f, "Recluster"),
- ClusterType::Hilbert => write!(f, "Hilbert Recluster"),
- },
+ MutationKind::Recluster => write!(f, "Recluster"),
  MutationKind::Update => write!(f, "Update"),
  MutationKind::Replace => write!(f, "Replace"),
  MutationKind::Compact => write!(f, "Compact"),

src/query/sql/src/executor/physical_plans/physical_recluster.rs

@@ -21,7 +21,6 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::schema::TableInfo;
 use databend_enterprise_hilbert_clustering::get_hilbert_clustering_handler;
-use databend_storages_common_table_meta::table::ClusterType;
 
 use crate::executor::physical_plans::CommitSink;
 use crate::executor::physical_plans::CompactSource;
@@ -90,7 +89,7 @@ impl PhysicalPlanBuilder {
  input: Box::new(plan),
  table_info,
  snapshot: Some(snapshot),
- mutation_kind: MutationKind::Recluster(ClusterType::Hilbert),
+ mutation_kind: MutationKind::Recluster,
  update_stream_meta: vec![],
  merge_meta: false,
  deduplicated_label: None,
@@ -139,7 +138,7 @@ impl PhysicalPlanBuilder {
  input: Box::new(root),
  table_info,
  snapshot: Some(snapshot),
- mutation_kind: MutationKind::Recluster(ClusterType::Linear),
+ mutation_kind: MutationKind::Recluster,
  update_stream_meta: vec![],
  merge_meta: false,
  deduplicated_label: None,

src/query/storages/fuse/src/operations/common/generators/mutation_generator.rs

@@ -101,7 +101,7 @@ impl SnapshotGenerator for MutationGenerator {
 
  if matches!(
  self.mutation_kind,
- MutationKind::Compact | MutationKind::Recluster(_)
+ MutationKind::Compact | MutationKind::Recluster
  ) {
  // for compaction, a basic but very important verification:
  // the number of rows should be the same

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

@@ -63,16 +63,17 @@ use crate::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 pub struct TableMutationAggregator {
  ctx: Arc<dyn TableContext>,
  schema: TableSchemaRef,
+ table_id: u64,
  dal: Operator,
  location_gen: TableMetaLocationGenerator,
-
  thresholds: BlockThresholds,
  block_per_seg: usize,
- default_cluster_key_id: Option<u32>,
 
+ default_cluster_key_id: Option<u32>,
  base_segments: Vec<Location>,
  // Used for recluster.
  recluster_merged_blocks: Vec<Arc<BlockMeta>>,
+ set_hilbert_level: bool,
 
  mutations: HashMap<SegmentIndex, BlockMutations>,
  appended_segments: Vec<Location>,
@@ -83,7 +84,6 @@ pub struct TableMutationAggregator {
  kind: MutationKind,
  start_time: Instant,
  finished_tasks: usize,
- table_id: u64,
 }
 
 // takes in table mutation logs and aggregates them (former mutation_transform)
@@ -115,7 +115,7 @@ impl AsyncAccumulatingTransform for TableMutationAggregator {
  },
  self.schema.clone(),
  )),
- MutationKind::Recluster(_) => self.apply_recluster(&mut new_segment_locs).await?,
+ MutationKind::Recluster => self.apply_recluster(&mut new_segment_locs).await?,
  _ => self.apply_mutation(&mut new_segment_locs).await?,
  };
 
@@ -137,13 +137,24 @@ impl TableMutationAggregator {
  removed_statistics: Statistics,
  kind: MutationKind,
  ) -> Self {
+ let set_hilbert_level = table
+ .cluster_type()
+ .is_some_and(|v| matches!(v, ClusterType::Hilbert))
+ && matches!(
+ kind,
+ MutationKind::Delete
+ | MutationKind::MergeInto
+ | MutationKind::Replace
+ | MutationKind::Recluster
+ );
  TableMutationAggregator {
  ctx,
  schema: table.schema(),
  dal: table.get_operator(),
  location_gen: table.meta_location_generator().clone(),
  thresholds: table.get_block_thresholds(),
  default_cluster_key_id: table.cluster_key_id(),
+ set_hilbert_level,
  block_per_seg: table
  .get_option(FUSE_OPT_KEY_BLOCK_PER_SEGMENT, DEFAULT_BLOCK_PER_SEGMENT),
  mutations: HashMap::new(),
@@ -262,6 +273,7 @@ impl TableMutationAggregator {
  let default_cluster_key = Some(default_cluster_key_id);
  let thresholds = self.thresholds;
  let block_per_seg = self.block_per_seg;
+ let set_hilbert_level = self.set_hilbert_level;
  let kind = self.kind;
  for chunk in &merged_blocks.into_iter().chunks(chunk_size) {
  let new_blocks = chunk.collect::<Vec<_>>();
@@ -279,6 +291,7 @@ impl TableMutationAggregator {
  all_perfect,
  block_per_seg,
  kind,
+ set_hilbert_level,
  )
  .await
  });
@@ -413,17 +426,19 @@ impl TableMutationAggregator {
  let thresholds = self.thresholds;
  let default_cluster_key_id = self.default_cluster_key_id;
  let block_per_seg = self.block_per_seg;
+ let kind = self.kind;
+ let set_hilbert_level = self.set_hilbert_level;
  let mut tasks = Vec::with_capacity(segment_indices.len());
  for index in segment_indices {
  let segment_mutation = self.mutations.remove(&index).unwrap();
  let location = self.base_segments.get(index).cloned();
  let schema = self.schema.clone();
  let op = self.dal.clone();
  let location_gen = self.location_gen.clone();
- let kind = self.kind;
 
- let mut all_perfect = false;
  tasks.push(async move {
+ let mut all_perfect = false;
+ let mut set_level = false;
  let (new_blocks, origin_summary) = if let Some(loc) = location {
  // read the old segment
  let compact_segment_info =
@@ -453,6 +468,12 @@ impl TableMutationAggregator {
 
  // assign back the mutated blocks to segment
  let new_blocks = block_editor.into_values().collect::<Vec<_>>();
+ set_level = set_hilbert_level
+ && segment_info
+ .summary
+ .cluster_stats
+ .as_ref()
+ .is_some_and(|v| v.cluster_key_id == default_cluster_key_id.unwrap());
  (new_blocks, Some(segment_info.summary))
  } else {
  // use by compact.
@@ -478,6 +499,7 @@ impl TableMutationAggregator {
  all_perfect,
  block_per_seg,
  kind,
+ set_level,
  )
  .await?;
 
@@ -551,6 +573,7 @@ async fn write_segment(
  all_perfect: bool,
  block_per_seg: usize,
  kind: MutationKind,
+ set_hilbert_level: bool,
 ) -> Result<(String, Statistics)> {
  let location = location_gen.gen_segment_info_location();
  let mut new_summary = reduce_block_metas(&blocks, thresholds, default_cluster_key);
@@ -564,9 +587,13 @@ async fn write_segment(
  new_summary.perfect_block_count = new_summary.block_count;
  }
  }
- if matches!(kind, MutationKind::Recluster(ClusterType::Hilbert)) {
- assert!(new_summary.cluster_stats.is_none());
- let level = if new_summary.block_count >= block_per_seg as u64 {
+ if set_hilbert_level {
+ debug_assert!(new_summary.cluster_stats.is_none());
+ let level = if new_summary.block_count >= block_per_seg as u64
+ && (new_summary.row_count as usize >= block_per_seg * thresholds.min_rows_per_block
+ || new_summary.uncompressed_byte_size as usize
+ >= block_per_seg * thresholds.max_bytes_per_block)
+ {
  -1
  } else {
  0

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

@@ -336,7 +336,7 @@ impl Processor for TransformSerializeBlock {
  }
  }
 
- if matches!(self.kind, MutationKind::Recluster(_)) {
+ if matches!(self.kind, MutationKind::Recluster) {
  Self::mutation_logs(MutationLogEntry::ReclusterAppendBlock {
  block_meta: Arc::new(block_meta),
  })