fix(fuse): isolate partition and cluster layouts

Author: KKould

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -280,9 +280,7 @@ async fn set_segment_format(
         match segment_format {
             FuseSegmentFormat::Row => unimplemented!(),
             FuseSegmentFormat::Column => {
-                let mut segment_builder =
-                    ColumnOrientedSegmentBuilder::new(table.schema().clone(), block_per_segment);
-                let segment_schema = segment_builder.segment_schema();
+                let segment_schema = table.schema();
                 let mut projection = HashSet::new();
                 for field in segment_schema.fields().iter() {
                     projection.insert(field.name.clone());
@@ -292,6 +290,10 @@ async fn set_segment_format(
                     .await?;
                 for segment in segments {
                     let segment = segment?;
+                    let mut segment_builder = ColumnOrientedSegmentBuilder::new(
+                        table.schema().clone(),
+                        block_per_segment,
+                    );
                     for block in segment.blocks {
                         segment_builder.add_block(block.as_ref().clone())?;
                     }

src/query/service/src/physical_plans/physical_commit_sink.rs

@@ -149,7 +149,7 @@ impl IPhysicalPlan for CommitSink {
             }
             CommitType::Mutation { kind, merge_meta } => {
                 if *merge_meta {
-                    let cluster_key_id = table.physical_cluster_key_id();
+                    let cluster_key_id = table.cluster_key_id();
                     builder.main_pipeline.add_accumulating_transformer(|| {
                         TransformMergeCommitMeta::create(cluster_key_id)
                     });

src/query/service/src/physical_plans/physical_compact_source.rs

@@ -97,7 +97,7 @@ impl IPhysicalPlan for CompactSource {
 
         let is_lazy = self.parts.partitions_type() == PartInfoType::LazyLevel;
         let thresholds = table.get_block_thresholds();
-        let cluster_key_id = table.physical_cluster_key_id();
+        let cluster_key_id = table.cluster_key_id();
         let partition_key_count = table.partition_key_count();
         let mut max_threads = builder.settings.get_max_threads()? as usize;
 

src/query/service/src/physical_plans/physical_multi_table_insert.rs

@@ -762,7 +762,7 @@ impl IPhysicalPlan for ChunkAppendData {
                 sort_builders.push(Box::new(builder.dummy_transform_builder()));
             }
             let partition_key_indices: Arc<[_]> =
-                cluster_stats_gen.cluster_key_index[..cluster_stats_gen.partition_key_count].into();
+                cluster_stats_gen.partition_key_index.clone().into();
             if !partition_key_indices.is_empty() {
                 partition_builders.push(Box::new(move |input, output| {
                     Ok(ProcessorPtr::create(AccumulatingTransformer::create(

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

@@ -119,7 +119,7 @@ pub async fn generate_segments_v2(
             TestFixture::default_table_meta_timestamps(),
         )
         .await?;
-        let mut summary = reduce_block_metas(&block_metas, BlockThresholds::default(), None);
+        let mut summary = reduce_block_metas(&block_metas, BlockThresholds::default(), None)?;
         let uuid = Uuid::new_v4();
         let location = format!(
             "{}/{}/{}_v{}.json",
@@ -162,7 +162,7 @@ pub async fn generate_segments(
             table_meta_timestamps,
         )
         .await?;
-        let mut summary = reduce_block_metas(&block_metas, BlockThresholds::default(), None);
+        let mut summary = reduce_block_metas(&block_metas, BlockThresholds::default(), None)?;
         let location = if is_greater_than_v5 {
             location_generator.gen_segment_info_location(table_meta_timestamps, false)
         } else {

src/query/service/tests/it/storages/fuse/bloom_index_meta_size.rs

@@ -331,6 +331,7 @@ fn build_test_segment_info(
         col_stats: col_stats.clone(),
         col_metas,
         cluster_stats: None,
+        partition_stats: None,
         location: block_location,
         bloom_filter_index_location: Some(location_gen.block_bloom_index_location(&block_uuid)),
         bloom_filter_index_size: 0,
@@ -368,6 +369,7 @@ fn build_test_segment_info(
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };

src/query/service/tests/it/storages/fuse/conflict.rs

@@ -105,6 +105,7 @@ fn test_resolvable_delete_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -133,6 +134,7 @@ fn test_resolvable_delete_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -154,6 +156,7 @@ fn test_resolvable_delete_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -175,6 +178,7 @@ fn test_resolvable_delete_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -219,6 +223,7 @@ fn test_resolvable_delete_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -260,6 +265,7 @@ fn test_resolvable_replace_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -288,6 +294,7 @@ fn test_resolvable_replace_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -309,6 +316,7 @@ fn test_resolvable_replace_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -330,6 +338,7 @@ fn test_resolvable_replace_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };
@@ -379,6 +388,7 @@ fn test_resolvable_replace_conflict() {
         virtual_col_stats: None,
         spatial_stats: None,
         cluster_stats: None,
+        partition_stats: None,
         virtual_block_count: None,
         additional_stats_meta: None,
     };

src/query/service/tests/it/storages/fuse/meta/column_oriented.rs

@@ -311,7 +311,7 @@ fn check_block_level_meta(
 }
 
 fn check_summary(block_metas: &[BlockMeta], column_oriented_segment: &ColumnOrientedSegment) {
-    let summary = reduce_block_metas(block_metas, Default::default(), Some(0));
+    let summary = reduce_block_metas(block_metas, Default::default(), Some(0)).unwrap();
     assert_eq!(summary.row_count, column_oriented_segment.summary.row_count);
     assert_eq!(
         summary.block_count,