diff --git a/src/query/service/src/physical_plans/physical_commit_sink.rs b/src/query/service/src/physical_plans/physical_commit_sink.rs
index fde0e57a43bf8..cac8d733ce4f1 100644
--- a/src/query/service/src/physical_plans/physical_commit_sink.rs
+++ b/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)
                     });
diff --git a/src/query/service/src/physical_plans/physical_compact_source.rs b/src/query/service/src/physical_plans/physical_compact_source.rs
index 53ffd233f46cd..a56d9ce7f751b 100644
--- a/src/query/service/src/physical_plans/physical_compact_source.rs
+++ b/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;
 
diff --git a/src/query/service/src/physical_plans/physical_multi_table_insert.rs b/src/query/service/src/physical_plans/physical_multi_table_insert.rs
index a8da67f856f56..303560a523a6d 100644
--- a/src/query/service/src/physical_plans/physical_multi_table_insert.rs
+++ b/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(
diff --git a/src/query/service/src/test_kits/fuse.rs b/src/query/service/src/test_kits/fuse.rs
index 224ea991e61d9..4dbbaed15cf62 100644
--- a/src/query/service/src/test_kits/fuse.rs
+++ b/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 {
diff --git a/src/query/service/tests/it/storages/fuse/bloom_index_meta_size.rs b/src/query/service/tests/it/storages/fuse/bloom_index_meta_size.rs
index eff08a15cba53..42056560328e9 100644
--- a/src/query/service/tests/it/storages/fuse/bloom_index_meta_size.rs
+++ b/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,
     };
diff --git a/src/query/service/tests/it/storages/fuse/conflict.rs b/src/query/service/tests/it/storages/fuse/conflict.rs
index 2768edcec610a..3b7b3cc712f6d 100644
--- a/src/query/service/tests/it/storages/fuse/conflict.rs
+++ b/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,
     };
diff --git a/src/query/service/tests/it/storages/fuse/meta/column_oriented.rs b/src/query/service/tests/it/storages/fuse/meta/column_oriented.rs
index 3759623a72b92..c24c302ea9252 100644
--- a/src/query/service/tests/it/storages/fuse/meta/column_oriented.rs
+++ b/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,
diff --git a/src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs b/src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs
index 22aed0ddd6da7..db065694d7531 100644
--- a/src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs
+++ b/src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs
@@ -59,6 +59,7 @@ use databend_storages_common_table_meta::meta::BlockMeta;
 use databend_storages_common_table_meta::meta::ClusterStatistics;
 use databend_storages_common_table_meta::meta::ColumnStatistics;
 use databend_storages_common_table_meta::meta::CompactSegmentInfo;
+use databend_storages_common_table_meta::meta::PartitionStatistics;
 use databend_storages_common_table_meta::meta::SegmentInfo;
 use databend_storages_common_table_meta::meta::VectorColumnStatistics;
 use databend_storages_common_table_meta::meta::VectorDistanceType;
@@ -106,6 +107,7 @@ fn new_test_mutator(
         1.0,
         thresholds,
         cluster_key_id,
+        0,
         max_tasks,
         mode,
         None,
@@ -167,6 +169,12 @@ fn make_recluster_block(
     ))
 }
 
+fn with_partition(mut block: Arc<BlockMeta>, partition: i32) -> Arc<BlockMeta> {
+    Arc::make_mut(&mut block).partition_stats =
+        Some(PartitionStatistics::new(vec![Scalar::from(partition)]));
+    block
+}
+
 async fn write_recluster_segment(
     data_accessor: &opendal::Operator,
     location_generator: &TableMetaLocationGenerator,
@@ -178,7 +186,7 @@ async fn write_recluster_segment(
         .iter()
         .map(|block| block.as_ref())
         .collect::<Vec<_>>();
-    let statistics = reduce_block_metas(&block_refs, thresholds, Some(cluster_key_id));
+    let statistics = reduce_block_metas(&block_refs, thresholds, Some(cluster_key_id))?;
     let segment = SegmentInfo::new(blocks, statistics);
     let segment_location = location_generator
         .gen_segment_info_location(TestFixture::default_table_meta_timestamps(), false);
@@ -339,6 +347,7 @@ async fn gen_recluster_segments_by_vector_stats(
     data_accessor: &opendal::Operator,
     location_generator: &TableMetaLocationGenerator,
     blocks_by_segment: &[VectorSegmentStatsSpec],
+    partitions: Option<&[i32]>,
     scalar_cluster_stats: bool,
     row_count: u64,
     block_size: u64,
@@ -346,8 +355,11 @@ async fn gen_recluster_segments_by_vector_stats(
     thresholds: BlockThresholds,
     cluster_key_id: u32,
 ) -> anyhow::Result<Vec<meta::Location>> {
+    if let Some(partitions) = partitions {
+        assert_eq!(partitions.len(), blocks_by_segment.len());
+    }
     let mut segment_locations = Vec::with_capacity(blocks_by_segment.len());
-    for blocks_spec in blocks_by_segment {
+    for (segment_index, blocks_spec) in blocks_by_segment.iter().enumerate() {
         let mut blocks = Vec::with_capacity(blocks_spec.len());
         for &(tenant_min, tenant_max, centroid, radius) in blocks_spec {
             let block_id = Uuid::new_v4().simple().to_string();
@@ -405,7 +417,12 @@ async fn gen_recluster_segments_by_vector_stats(
                 Some(Utc::now()),
             );
             block.vector_stats = Some(vector_stats);
-            blocks.push(Arc::new(block));
+            let block = Arc::new(block);
+            blocks.push(if let Some(partitions) = partitions {
+                with_partition(block, partitions[segment_index])
+            } else {
+                block
+            });
         }
 
         segment_locations.push(
@@ -557,6 +574,28 @@ fn task_part_counts(parts: &ReclusterParts) -> Vec<usize> {
         .collect::<Vec<_>>()
 }
 
+fn assert_partition_isolated(windows: &[Vec<SelectedReclusterSegment>], expected_segments: usize) {
+    let mut selected_segments = HashSet::new();
+    for window in windows {
+        let partitions = window
+            .iter()
+            .map(|segment| {
+                selected_segments.insert(segment.loc.segment_idx);
+                segment
+                    .info
+                    .summary
+                    .partition_stats
+                    .as_ref()
+                    .unwrap()
+                    .values
+                    .clone()
+            })
+            .collect::<HashSet<_>>();
+        assert_eq!(partitions.len(), 1);
+    }
+    assert_eq!(selected_segments.len(), expected_segments);
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_recluster_limit_skips_empty_range() -> anyhow::Result<()> {
     let fixture = TestFixture::setup().await?;
@@ -1073,6 +1112,64 @@ async fn test_select_segments_covers_candidates() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_scalar_segment_selection_does_not_cross_partitions() -> anyhow::Result<()> {
+    let fixture = TestFixture::setup().await?;
+    let ctx = fixture.new_query_ctx().await?;
+    let data_accessor = ctx.get_application_level_data_operator()?.operator();
+    let location_generator = TableMetaLocationGenerator::new("_prefix".to_owned());
+    let cluster_key_id = 0;
+    let thresholds = BlockThresholds::new(1000, 100, 100, 1);
+
+    let mut segment_locations = Vec::new();
+    for (partition, min, max) in [(1, 1, 10), (2, 1, 10), (1, 2, 9), (2, 2, 9)] {
+        let block = with_partition(
+            make_recluster_block(cluster_key_id, min, max, 0, 1000, 100, 100),
+            partition,
+        );
+        segment_locations.push(
+            write_recluster_segment(
+                &data_accessor,
+                &location_generator,
+                vec![block],
+                thresholds,
+                cluster_key_id,
+            )
+            .await?,
+        );
+    }
+
+    let schema = test_cluster_schema();
+    let ctx: Arc<dyn TableContext> = ctx;
+    let compact_segments = segment_pruning(
+        &ctx,
+        schema.clone(),
+        data_accessor.clone(),
+        create_segment_location_vector(segment_locations, None),
+    )
+    .await?;
+    let mutator = ReclusterMutator::new(
+        ctx,
+        data_accessor,
+        schema,
+        test_cluster_key_exprs(),
+        1.0,
+        thresholds,
+        cluster_key_id,
+        1,
+        1,
+        ReclusterMode::Aggressive,
+        None,
+    );
+
+    let windows = mutator.select_segments(&compact_segments, 8)?;
+    assert_eq!(windows.len(), 2);
+    assert!(windows.iter().all(|window| window.len() == 2));
+    assert_partition_isolated(&windows, 4);
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_select_segments_normal_conservative() -> anyhow::Result<()> {
     let fixture = TestFixture::setup().await?;
@@ -1202,6 +1299,7 @@ async fn test_recluster_mutator_vector_mixed_key_overlap_selection() -> anyhow::
             // Scalar range overlaps with segment 0/1, but vector sphere does not.
             vec![(1, 1, [10.0, 0.0], 1.0)],
         ],
+        None,
         true,
         1000,
         100,
@@ -1236,6 +1334,7 @@ async fn test_recluster_mutator_vector_mixed_key_overlap_selection() -> anyhow::
         1.0,
         thresholds,
         cluster_key_id,
+        0,
         1,
         ReclusterMode::Conservative,
         Some(vector_cluster_info),
@@ -1280,6 +1379,7 @@ async fn test_recluster_mutator_vector_only_overlap_selection() -> anyhow::Resul
             vec![(2, 2, [0.5, 0.0], 1.0)],
             vec![(3, 3, [10.0, 0.0], 1.0)],
         ],
+        None,
         false,
         1000,
         100,
@@ -1314,6 +1414,7 @@ async fn test_recluster_mutator_vector_only_overlap_selection() -> anyhow::Resul
         1.0,
         thresholds,
         cluster_key_id,
+        0,
         1,
         ReclusterMode::Conservative,
         Some(vector_cluster_info),
@@ -1338,6 +1439,74 @@ async fn test_recluster_mutator_vector_only_overlap_selection() -> anyhow::Resul
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_vector_segment_selection_does_not_cross_partitions() -> anyhow::Result<()> {
+    let fixture = TestFixture::setup().await?;
+    let ctx = fixture.new_query_ctx().await?;
+    ctx.get_settings().set_recluster_block_size(1000)?;
+
+    let data_accessor = ctx.get_application_level_data_operator()?.operator();
+    let location_generator = TableMetaLocationGenerator::new("_prefix".to_owned());
+    let cluster_key_id = 0;
+    let thresholds = BlockThresholds::new(1000, 100, 100, 1);
+    let segment_locations = gen_recluster_segments_by_vector_stats(
+        &data_accessor,
+        &location_generator,
+        &[
+            vec![(1, 1, [0.0, 0.0], 1.0)],
+            vec![(1, 1, [0.5, 0.0], 1.0)],
+            // This sphere overlaps segments 0 and 1, but belongs to another partition.
+            vec![(1, 1, [0.25, 0.0], 1.0)],
+            // These two are handled by the non-overlap window path.
+            vec![(1, 1, [100.0, 0.0], 1.0)],
+            vec![(1, 1, [200.0, 0.0], 1.0)],
+        ],
+        Some(&[1, 1, 2, 1, 2]),
+        false,
+        1000,
+        100,
+        100,
+        thresholds,
+        cluster_key_id,
+    )
+    .await?;
+
+    let schema = vector_recluster_schema();
+    let ctx: Arc<dyn TableContext> = ctx;
+    let compact_segments = segment_pruning(
+        &ctx,
+        schema.clone(),
+        data_accessor.clone(),
+        create_segment_location_vector(segment_locations, None),
+    )
+    .await?;
+    let vector_cluster_info = VectorClusterInfo {
+        key_index: 0,
+        column_id: 1,
+        column_name: "embedding".to_string(),
+        dimension: 2,
+        distance_type: VectorDistanceType::L2,
+    };
+    let mutator = ReclusterMutator::new(
+        ctx,
+        data_accessor,
+        schema,
+        vec![test_vector_cluster_key_expr()],
+        1.0,
+        thresholds,
+        cluster_key_id,
+        1,
+        1,
+        ReclusterMode::Aggressive,
+        Some(vector_cluster_info),
+    );
+
+    let windows = mutator.select_segments(&compact_segments, 8)?;
+    assert_partition_isolated(&windows, 5);
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_accumulates_tasks_across_windows() -> anyhow::Result<()> {
     let fixture = TestFixture::setup().await?;
diff --git a/src/query/service/tests/it/storages/fuse/statistics.rs b/src/query/service/tests/it/storages/fuse/statistics.rs
index 6b87320fcdb73..42efabd57d7dc 100644
--- a/src/query/service/tests/it/storages/fuse/statistics.rs
+++ b/src/query/service/tests/it/storages/fuse/statistics.rs
@@ -883,7 +883,7 @@ fn char_len(value: &str) -> usize {
 fn test_reduce_block_meta() -> anyhow::Result<()> {
     // case 1: empty input should return the default statistics
     let block_metas: Vec<BlockMeta> = vec![];
-    let reduced = reduce_block_metas(&block_metas, BlockThresholds::default(), None);
+    let reduced = reduce_block_metas(&block_metas, BlockThresholds::default(), None)?;
     assert_eq!(Statistics::default(), reduced);
 
     // case 2: accumulated variants of size index should be as expected
@@ -929,7 +929,7 @@ fn test_reduce_block_meta() -> anyhow::Result<()> {
         blocks.push(block_meta);
     }
 
-    let stats = reduce_block_metas(&blocks, BlockThresholds::default(), None);
+    let stats = reduce_block_metas(&blocks, BlockThresholds::default(), None)?;
 
     assert_eq!(acc_row_count, stats.row_count);
     assert_eq!(acc_block_size, stats.uncompressed_byte_size);
diff --git a/src/query/storages/common/cache/src/manager.rs b/src/query/storages/common/cache/src/manager.rs
index 39914f449644c..228d95dc188fa 100644
--- a/src/query/storages/common/cache/src/manager.rs
+++ b/src/query/storages/common/cache/src/manager.rs
@@ -1298,6 +1298,7 @@ mod tests {
             col_stats: Default::default(),
             col_metas: Default::default(),
             cluster_stats: None,
+            partition_stats: None,
             location: ("".to_string(), 0),
             bloom_filter_index_location: None,
             bloom_filter_index_size: 0,
diff --git a/src/query/storages/common/table_meta/src/meta/column_oriented_segment/segment_builder.rs b/src/query/storages/common/table_meta/src/meta/column_oriented_segment/segment_builder.rs
index ad059421063d0..6952d420a78d2 100644
--- a/src/query/storages/common/table_meta/src/meta/column_oriented_segment/segment_builder.rs
+++ b/src/query/storages/common/table_meta/src/meta/column_oriented_segment/segment_builder.rs
@@ -46,10 +46,12 @@ use crate::meta::ClusterStatistics;
 use crate::meta::ColumnStatistics;
 use crate::meta::Location;
 use crate::meta::MetaEncoding;
+use crate::meta::PartitionStatistics;
 use crate::meta::Statistics;
 use crate::meta::VirtualBlockMeta;
 use crate::meta::format::encode;
 use crate::meta::supported_stat_type;
+use crate::meta::validate_segment_partition_statistics;
 
 pub trait SegmentBuilder: Send + Sync + 'static {
     type Segment: AbstractSegment;
@@ -69,6 +71,7 @@ pub struct ColumnOrientedSegmentBuilder {
     block_size: Vec<u64>,
     file_size: Vec<u64>,
     cluster_stats: Vec<Option<ClusterStatistics>>,
+    partition_stats: Vec<Option<PartitionStatistics>>,
     location: (Vec<String>, Vec<u64>),
     bloom_filter_index_location: LocationsWithOption,
     bloom_filter_index_size: Vec<u64>,
@@ -128,6 +131,7 @@ impl SegmentBuilder for ColumnOrientedSegmentBuilder {
         self.block_size.push(block_meta.block_size);
         self.file_size.push(block_meta.file_size);
         self.cluster_stats.push(block_meta.cluster_stats);
+        self.partition_stats.push(block_meta.partition_stats);
         self.location.0.push(block_meta.location.0);
         self.location.1.push(block_meta.location.1);
         self.bloom_filter_index_location
@@ -250,6 +254,7 @@ impl SegmentBuilder for ColumnOrientedSegmentBuilder {
             block_size: Vec::with_capacity(block_per_segment),
             file_size: Vec::with_capacity(block_per_segment),
             cluster_stats: Vec::with_capacity(block_per_segment),
+            partition_stats: Vec::with_capacity(block_per_segment),
             location: (
                 Vec::with_capacity(block_per_segment),
                 Vec::with_capacity(block_per_segment),
@@ -351,6 +356,8 @@ impl ColumnOrientedSegmentBuilder {
         self.column_stats = self_column_stats;
 
         let cluster_stats = reduce_cluster_statistics(&self.cluster_stats, default_cluster_key_id);
+        let partition_stats =
+            validate_segment_partition_statistics(self.partition_stats.iter().map(Option::as_ref))?;
 
         Ok(Statistics {
             row_count,
@@ -369,6 +376,7 @@ impl ColumnOrientedSegmentBuilder {
             virtual_col_stats: None,
             spatial_stats: None,
             cluster_stats,
+            partition_stats,
             virtual_block_count: Some(virtual_block_count),
             additional_stats_meta,
         })
diff --git a/src/query/storages/common/table_meta/src/meta/current/mod.rs b/src/query/storages/common/table_meta/src/meta/current/mod.rs
index 4e0719702e9e9..e000a3ad04b62 100644
--- a/src/query/storages/common/table_meta/src/meta/current/mod.rs
+++ b/src/query/storages/common/table_meta/src/meta/current/mod.rs
@@ -21,6 +21,7 @@ pub use v2::ColumnStatistics;
 pub use v2::DraftVirtualBlockMeta;
 pub use v2::DraftVirtualColumnMeta;
 pub use v2::ExtendedBlockMeta;
+pub use v2::PartitionStatistics;
 pub use v2::SegmentStatistics;
 pub use v2::SpatialStatistics;
 pub use v2::Statistics;
@@ -28,6 +29,7 @@ pub use v2::VectorColumnStatistics;
 pub use v2::VectorDistanceType;
 pub use v2::VirtualBlockMeta;
 pub use v2::VirtualColumnMeta;
+pub use v2::validate_segment_partition_statistics;
 pub use v4::CompactSegmentInfo;
 pub use v4::RawBlockMeta;
 pub use v4::SegmentInfo;
diff --git a/src/query/storages/common/table_meta/src/meta/v2/mod.rs b/src/query/storages/common/table_meta/src/meta/v2/mod.rs
index 8db7b00c17bf3..285afd107467c 100644
--- a/src/query/storages/common/table_meta/src/meta/v2/mod.rs
+++ b/src/query/storages/common/table_meta/src/meta/v2/mod.rs
@@ -31,8 +31,10 @@ pub use snapshot::TableSnapshot;
 pub use statistics::AdditionalStatsMeta;
 pub use statistics::ClusterStatistics;
 pub use statistics::ColumnStatistics;
+pub use statistics::PartitionStatistics;
 pub use statistics::SpatialStatistics;
 pub use statistics::Statistics;
 pub use statistics::VectorColumnStatistics;
 pub use statistics::VectorDistanceType;
+pub use statistics::validate_segment_partition_statistics;
 pub use table_snapshot_statistics::TableSnapshotStatistics;
diff --git a/src/query/storages/common/table_meta/src/meta/v2/segment.rs b/src/query/storages/common/table_meta/src/meta/v2/segment.rs
index cd2f594366c46..8967f63e0158f 100644
--- a/src/query/storages/common/table_meta/src/meta/v2/segment.rs
+++ b/src/query/storages/common/table_meta/src/meta/v2/segment.rs
@@ -38,6 +38,7 @@ use crate::meta::ColumnStatistics;
 use crate::meta::Compression;
 use crate::meta::FormatVersion;
 use crate::meta::Location;
+use crate::meta::PartitionStatistics;
 use crate::meta::SpatialStatistics;
 use crate::meta::Statistics;
 use crate::meta::StatisticsOfVectorColumns;
@@ -182,6 +183,8 @@ pub struct BlockMeta {
     pub col_stats: HashMap<ColumnId, ColumnStatistics>,
     pub col_metas: HashMap<ColumnId, ColumnMeta>,
     pub cluster_stats: Option<ClusterStatistics>,
+    #[serde(default)]
+    pub partition_stats: Option<PartitionStatistics>,
     /// location of data block
     pub location: Location,
     /// location of bloom filter index
@@ -235,6 +238,7 @@ impl BlockMeta {
             col_stats,
             col_metas,
             cluster_stats,
+            partition_stats: None,
             location,
             bloom_filter_index_location,
             bloom_filter_index_size,
@@ -369,6 +373,7 @@ impl BlockMeta {
             col_stats,
             col_metas,
             cluster_stats: None,
+            partition_stats: None,
             location: (s.location.path.clone(), 0),
             bloom_filter_index_location: None,
             bloom_filter_index_size: 0,
@@ -401,6 +406,7 @@ impl BlockMeta {
             col_stats,
             col_metas,
             cluster_stats: None,
+            partition_stats: None,
             location: s.location.clone(),
             bloom_filter_index_location: s.bloom_filter_index_location.clone(),
             bloom_filter_index_size: s.bloom_filter_index_size,
diff --git a/src/query/storages/common/table_meta/src/meta/v2/statistics.rs b/src/query/storages/common/table_meta/src/meta/v2/statistics.rs
index 31246c5e26f74..b3cac6946773a 100644
--- a/src/query/storages/common/table_meta/src/meta/v2/statistics.rs
+++ b/src/query/storages/common/table_meta/src/meta/v2/statistics.rs
@@ -17,6 +17,7 @@ use std::fmt;
 use std::marker::PhantomData;
 
 use databend_common_base::base::OrderedFloat;
+use databend_common_exception::ErrorCode;
 use databend_common_expression::ColumnId;
 use databend_common_expression::Scalar;
 use databend_common_expression::TableDataType;
@@ -80,6 +81,46 @@ pub struct ClusterStatistics {
     pub pages: Option<Vec<Scalar>>,
 }
 
+/// Exact values of the PARTITION BY expressions for a block or segment.
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, FrozenAPI)]
+pub struct PartitionStatistics {
+    #[serde(
+        serialize_with = "serialize_index_scalar_vec",
+        deserialize_with = "deserialize_index_scalar_vec"
+    )]
+    pub values: Vec<Scalar>,
+}
+
+impl PartitionStatistics {
+    pub fn new(values: Vec<Scalar>) -> Self {
+        Self { values }
+    }
+}
+
+pub fn validate_segment_partition_statistics<'a>(
+    stats: impl IntoIterator<Item = Option<&'a PartitionStatistics>>,
+) -> databend_common_exception::Result<Option<PartitionStatistics>> {
+    let mut partition = None;
+    let mut has_unknown = false;
+    for stats in stats {
+        match (partition, stats) {
+            (Some(expected), Some(actual)) if expected != actual => {
+                return Err(ErrorCode::Internal(
+                    "segment contains blocks from different partitions",
+                ));
+            }
+            (None, Some(actual)) => partition = Some(actual),
+            (_, None) => has_unknown = true,
+            _ => {}
+        }
+    }
+    if has_unknown {
+        Ok(None)
+    } else {
+        Ok(partition.cloned())
+    }
+}
+
 /// Spatial statistics for geometry columns.
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, Default, FrozenAPI)]
 pub struct SpatialStatistics {
@@ -268,6 +309,8 @@ pub struct Statistics {
     pub virtual_col_stats: Option<HashMap<ColumnId, ColumnStatistics>>,
     pub spatial_stats: Option<HashMap<ColumnId, SpatialStatistics>>,
     pub cluster_stats: Option<ClusterStatistics>,
+    #[serde(default)]
+    pub partition_stats: Option<PartitionStatistics>,
     pub virtual_block_count: Option<u64>,
 
     pub additional_stats_meta: Option<AdditionalStatsMeta>,
@@ -432,6 +475,7 @@ impl Statistics {
             virtual_col_stats: None,
             spatial_stats: None,
             cluster_stats: None,
+            partition_stats: None,
             virtual_block_count: None,
             additional_stats_meta: None,
         }
@@ -663,4 +707,20 @@ mod tests {
 
         assert_eq!(decoded, ClusterStatistics::new(7, stats.min, stats.max, 2));
     }
+
+    #[test]
+    fn segment_partition_statistics_rejects_different_partitions() {
+        let left = PartitionStatistics::new(vec![Scalar::Number(1_i64.into())]);
+        let right = PartitionStatistics::new(vec![Scalar::Number(2_i64.into())]);
+
+        assert_eq!(
+            validate_segment_partition_statistics([Some(&left), Some(&left)]).unwrap(),
+            Some(left.clone())
+        );
+        assert!(validate_segment_partition_statistics([Some(&left), Some(&right)]).is_err());
+        assert_eq!(
+            validate_segment_partition_statistics([None, Some(&left)]).unwrap(),
+            None
+        );
+    }
 }
diff --git a/src/query/storages/common/table_meta/src/meta/v3/frozen/block_meta.rs b/src/query/storages/common/table_meta/src/meta/v3/frozen/block_meta.rs
index f5034e796a578..8e7bd877a6d62 100644
--- a/src/query/storages/common/table_meta/src/meta/v3/frozen/block_meta.rs
+++ b/src/query/storages/common/table_meta/src/meta/v3/frozen/block_meta.rs
@@ -58,6 +58,7 @@ impl From<BlockMeta> for crate::meta::BlockMeta {
                 .map(|(k, v)| (k, v.into()))
                 .collect(),
             cluster_stats: value.cluster_stats.map(|v| v.into()),
+            partition_stats: None,
             location: value.location,
             bloom_filter_index_location: value.bloom_filter_index_location,
             bloom_filter_index_size: value.bloom_filter_index_size,
diff --git a/src/query/storages/common/table_meta/src/meta/v3/frozen/statistics.rs b/src/query/storages/common/table_meta/src/meta/v3/frozen/statistics.rs
index b84d4e50df9c4..e03f0569c70cd 100644
--- a/src/query/storages/common/table_meta/src/meta/v3/frozen/statistics.rs
+++ b/src/query/storages/common/table_meta/src/meta/v3/frozen/statistics.rs
@@ -56,6 +56,7 @@ impl From<Statistics> for crate::meta::Statistics {
             virtual_col_stats: None,
             spatial_stats: None,
             cluster_stats: None,
+            partition_stats: None,
             virtual_block_count: None,
             additional_stats_meta: None,
         }
diff --git a/src/query/storages/common/table_meta/src/meta/v4/segment.rs b/src/query/storages/common/table_meta/src/meta/v4/segment.rs
index 65dbf75b4bfd9..59a042ce8d9d0 100644
--- a/src/query/storages/common/table_meta/src/meta/v4/segment.rs
+++ b/src/query/storages/common/table_meta/src/meta/v4/segment.rs
@@ -363,6 +363,7 @@ mod tests {
     use crate::meta::ColumnMeta;
     use crate::meta::ColumnStatistics;
     use crate::meta::Compression;
+    use crate::meta::PartitionStatistics;
 
     #[allow(dead_code)]
     #[derive(Serialize)]
@@ -546,6 +547,9 @@ mod tests {
             col_stats: block_col_stats,
             col_metas,
             cluster_stats: None,
+            partition_stats: Some(PartitionStatistics::new(vec![Scalar::String(
+                "partition-a".to_string(),
+            )])),
             location: ("block.parquet".to_string(), 0),
             bloom_filter_index_location: None,
             bloom_filter_index_size: 0,
@@ -570,6 +574,9 @@ mod tests {
             compressed_byte_size: 16,
             index_size: 0,
             col_stats: summary_col_stats,
+            partition_stats: Some(PartitionStatistics::new(vec![Scalar::String(
+                "partition-a".to_string(),
+            )])),
             ..Default::default()
         })
     }
@@ -586,8 +593,10 @@ mod tests {
             Scalar::String("aaa".to_string())
         );
         assert_eq!(segment.blocks.len(), 1);
+        assert!(segment.summary.partition_stats.is_none());
 
         let block = &segment.blocks[0];
+        assert!(block.partition_stats.is_none());
         assert_eq!(block.row_count, 3);
         assert_eq!(block.location.0, "block.parquet");
         assert!(block.col_metas[&1].as_parquet().is_some());
@@ -636,6 +645,14 @@ mod tests {
         );
 
         assert_eq!(segment.blocks.len(), 1);
+        assert_eq!(
+            segment.summary.partition_stats.as_ref().unwrap().values,
+            vec![Scalar::String("partition-a".to_string())]
+        );
+        assert_eq!(
+            segment.blocks[0].partition_stats.as_ref().unwrap().values,
+            vec![Scalar::String("partition-a".to_string())]
+        );
         assert_eq!(
             segment.blocks[0].col_stats[&3].max,
             Scalar::Decimal(DecimalScalar::Decimal64(
diff --git a/src/query/storages/fuse/src/fuse_table.rs b/src/query/storages/fuse/src/fuse_table.rs
index 3141f8c31af6b..1e637a770889b 100644
--- a/src/query/storages/fuse/src/fuse_table.rs
+++ b/src/query/storages/fuse/src/fuse_table.rs
@@ -614,57 +614,43 @@ impl FuseTable {
         self.resolve_partition_keys().map_or(0, |keys| keys.len())
     }
 
-    /// The key used for physical Fuse layout. Partition expressions are always the
-    /// leading dimensions, followed by the user-visible linear cluster key.
-    pub fn resolve_physical_cluster_keys(&self) -> Option<Vec<AstExpr>> {
-        let mut keys = self.resolve_partition_keys().unwrap_or_default();
-        if let Some(cluster_keys) = self.resolve_cluster_keys() {
-            keys.extend(cluster_keys);
-        }
-        (!keys.is_empty()).then_some(keys)
-    }
-
-    pub fn physical_cluster_key_id(&self) -> Option<u32> {
-        self.cluster_key_id()
-            .or_else(|| self.partition_key_str().map(|_| 0))
-    }
-
     pub fn partition_pruning_info(
         &self,
         ctx: Arc<dyn TableContext>,
     ) -> Option<PartitionPruningInfo> {
-        let partition_key_count = self.partition_key_count();
-        if partition_key_count == 0 {
-            return None;
-        }
-        Some(PartitionPruningInfo {
-            cluster_key_id: self.physical_cluster_key_id().unwrap(),
-            partition_keys: self
-                .linear_cluster_keys(ctx)
-                .into_iter()
-                .take(partition_key_count)
-                .collect(),
-        })
+        let partition_keys = self.linear_partition_keys(ctx);
+        (!partition_keys.is_empty()).then_some(PartitionPruningInfo { partition_keys })
+    }
+
+    pub fn linear_partition_keys(&self, ctx: Arc<dyn TableContext>) -> Vec<RemoteExpr<String>> {
+        self.linear_keys(ctx, self.resolve_partition_keys())
     }
 
     pub fn linear_cluster_keys(&self, ctx: Arc<dyn TableContext>) -> Vec<RemoteExpr<String>> {
-        let Some(cluster_key_exprs) = self.resolve_physical_cluster_keys() else {
+        self.linear_keys(ctx, self.resolve_cluster_keys())
+    }
+
+    fn linear_keys(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        key_exprs: Option<Vec<AstExpr>>,
+    ) -> Vec<RemoteExpr<String>> {
+        let Some(key_exprs) = key_exprs else {
             return vec![];
         };
 
         let table_meta = Arc::new(self.clone());
-        let exprs = parse_cluster_keys(ctx, table_meta.clone(), cluster_key_exprs).unwrap();
-        let cluster_keys = exprs
+        parse_cluster_keys(ctx, table_meta.clone(), key_exprs)
+            .unwrap()
             .iter()
-            .map(|k| {
-                k.project_column_ref(|index| {
+            .map(|key| {
+                key.project_column_ref(|index| {
                     Ok(table_meta.schema().field(*index).name().to_string())
                 })
                 .unwrap()
                 .as_remote_expr()
             })
-            .collect();
-        cluster_keys
+            .collect()
     }
 
     pub fn bloom_index_cols(&self) -> BloomIndexColumns {
@@ -687,7 +673,7 @@ impl FuseTable {
     }
 
     pub fn cluster_key_types(&self, ctx: Arc<dyn TableContext>) -> Vec<DataType> {
-        let Some(ast_exprs) = self.resolve_physical_cluster_keys() else {
+        let Some(ast_exprs) = self.resolve_cluster_keys() else {
             return vec![];
         };
         let cluster_keys = parse_cluster_keys(ctx, Arc::new(self.clone()), ast_exprs).unwrap();
@@ -991,7 +977,8 @@ impl FuseTable {
     pub fn enable_stream_block_write(&self, ctx: Arc<dyn TableContext>) -> Result<bool> {
         Ok(ctx.get_settings().get_enable_block_stream_write()?
             && matches!(self.storage_format, FuseStorageFormat::Parquet)
-            && self.resolve_physical_cluster_keys().is_none())
+            && self.cluster_key_meta().is_none()
+            && self.partition_key_count() == 0)
     }
 
     pub fn with_schema(&self, schema: Arc<TableSchema>) -> Arc<FuseTable> {
diff --git a/src/query/storages/fuse/src/io/write/block_writer.rs b/src/query/storages/fuse/src/io/write/block_writer.rs
index 478d7807fb483..02d4364b07147 100644
--- a/src/query/storages/fuse/src/io/write/block_writer.rs
+++ b/src/query/storages/fuse/src/io/write/block_writer.rs
@@ -167,6 +167,9 @@ impl BlockBuilder {
     pub fn build<F>(&self, data_block: DataBlock, f: F) -> Result<BlockSerialization>
     where F: Fn(DataBlock, &ClusterStatsGenerator) -> Result<(Option<ClusterStatistics>, DataBlock)>
     {
+        let partition_stats = self
+            .cluster_stats_gen
+            .extract_partition_stats(&data_block)?;
         let (cluster_stats, data_block) = f(data_block, &self.cluster_stats_gen)?;
         let (block_location, block_id) = self
             .meta_locations
@@ -285,6 +288,7 @@ impl BlockBuilder {
             col_stats,
             col_metas,
             cluster_stats,
+            partition_stats,
             location: block_location,
             bloom_filter_index_location: bloom_index_state.as_ref().map(|v| v.location.clone()),
             bloom_filter_index_size: bloom_index_state
diff --git a/src/query/storages/fuse/src/io/write/stream/block_builder.rs b/src/query/storages/fuse/src/io/write/stream/block_builder.rs
index 91b47288a6d7b..d2ab9728fddd6 100644
--- a/src/query/storages/fuse/src/io/write/stream/block_builder.rs
+++ b/src/query/storages/fuse/src/io/write/stream/block_builder.rs
@@ -406,6 +406,7 @@ impl StreamBlockBuilder {
             // Stream block writing is only enabled for tables without a cluster key, so cluster
             // statistics cannot be produced on this path.
             cluster_stats: None,
+            partition_stats: None,
             location: block_location,
             bloom_filter_index_location: bloom_index_state.as_ref().map(|v| v.location.clone()),
             bloom_filter_index_size: bloom_index_state
diff --git a/src/query/storages/fuse/src/operations/analyze/analyze_state_sink.rs b/src/query/storages/fuse/src/operations/analyze/analyze_state_sink.rs
index 61b81ad4793c4..819bfd68457b8 100644
--- a/src/query/storages/fuse/src/operations/analyze/analyze_state_sink.rs
+++ b/src/query/storages/fuse/src/operations/analyze/analyze_state_sink.rs
@@ -550,7 +550,7 @@ impl SinkAnalyzeState {
         Option<ClusterStatistics>,
     )> {
         // 1. Read table snapshot.
-        let default_cluster_key_id = table.physical_cluster_key_id();
+        let default_cluster_key_id = table.cluster_key_id();
 
         // 2. Iterator segments and blocks to estimate statistics.
         let mut read_segment_count = 0;
diff --git a/src/query/storages/fuse/src/operations/append.rs b/src/query/storages/fuse/src/operations/append.rs
index c65d75b5f556d..cd2540a841d60 100644
--- a/src/query/storages/fuse/src/operations/append.rs
+++ b/src/query/storages/fuse/src/operations/append.rs
@@ -172,8 +172,7 @@ impl FuseTable {
             }
             pipeline.add_pipe(builder.finalize());
         }
-        let partition_key_indices: Arc<[_]> =
-            cluster_stats_gen.cluster_key_index[..cluster_stats_gen.partition_key_count].into();
+        let partition_key_indices: Arc<[_]> = cluster_stats_gen.partition_key_index.clone().into();
         if !partition_key_indices.is_empty() {
             let mut builder = pipeline.add_transform_with_specified_len(
                 move |input, output| {
@@ -255,8 +254,7 @@ impl FuseTable {
                 move || TransformSortPartial::new(LimitType::None, sort_desc.clone())
             });
         }
-        let partition_key_indices: Arc<[_]> =
-            cluster_stats_gen.cluster_key_index[..cluster_stats_gen.partition_key_count].into();
+        let partition_key_indices: Arc<[_]> = cluster_stats_gen.partition_key_index.clone().into();
         if !partition_key_indices.is_empty() {
             if rewrite_replaced_block {
                 pipeline.add_accumulating_transformer(move || {
@@ -278,25 +276,34 @@ impl FuseTable {
         block_thresholds: BlockThresholds,
         input_schema: Arc<DataSchema>,
     ) -> Result<ClusterStatsGenerator> {
-        if self.physical_cluster_key_id().is_none() {
+        let partition_keys = self.linear_partition_keys(ctx.clone());
+        let cluster_keys = self.linear_cluster_keys(ctx.clone());
+        if partition_keys.is_empty() && cluster_keys.is_empty() {
             return Ok(ClusterStatsGenerator::default());
         }
 
         let mut merged = input_schema.fields().clone();
-
-        let cluster_keys = self.linear_cluster_keys(ctx.clone());
+        let mut partition_key_index = Vec::with_capacity(partition_keys.len());
         let mut cluster_key_index = Vec::with_capacity(cluster_keys.len());
         let mut extra_key_num = 0;
 
-        let mut exprs = Vec::with_capacity(cluster_keys.len());
+        let mut exprs = Vec::with_capacity(partition_keys.len() + cluster_keys.len());
         let mut vector_cluster_info = None;
         let mut vector_column_input_offset = None;
 
-        for (key_index, remote_expr) in cluster_keys.iter().enumerate() {
+        let keys = partition_keys.iter().map(|expr| (None, expr)).chain(
+            cluster_keys
+                .iter()
+                .enumerate()
+                .map(|(index, expr)| (Some(index), expr)),
+        );
+        for (cluster_key_position, remote_expr) in keys {
             let expr = remote_expr
                 .as_expr(&BUILTIN_FUNCTIONS)
                 .project_column_ref(|name| input_schema.index_of(name))?;
-            if let DataType::Vector(vector_ty) = expr.data_type().remove_nullable() {
+            if let (Some(key_index), DataType::Vector(vector_ty)) =
+                (cluster_key_position, expr.data_type().remove_nullable())
+            {
                 let Expr::ColumnRef(ColumnRef { id, .. }) = &expr else {
                     return Err(ErrorCode::InvalidClusterKeys(
                         "Vector cluster key only supports direct column reference",
@@ -342,7 +349,11 @@ impl FuseTable {
                     offset
                 }
             };
-            cluster_key_index.push(index);
+            if cluster_key_position.is_some() {
+                cluster_key_index.push(index);
+            } else {
+                partition_key_index.push(index);
+            }
         }
 
         let operators = if exprs.is_empty() {
@@ -377,7 +388,7 @@ impl FuseTable {
         }
 
         let mut generator = ClusterStatsGenerator::new(
-            self.physical_cluster_key_id().unwrap(),
+            self.cluster_key_id().unwrap_or(0),
             cluster_key_index,
             extra_key_num,
             level,
@@ -387,7 +398,7 @@ impl FuseTable {
             merged,
             ctx.get_function_context()?,
         );
-        generator.partition_key_count = self.partition_key_count();
+        generator.partition_key_index = partition_key_index;
         Ok(generator)
     }
 
diff --git a/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs b/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs
index a97aa6e9fe394..a6b15a653b502 100644
--- a/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs
+++ b/src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs
@@ -306,7 +306,7 @@ impl AsyncSink for CommitMultiTableInsert {
                 *m = TransformMergeCommitMeta::merge_commit_meta(
                     m.clone(),
                     meta,
-                    table.physical_cluster_key_id(),
+                    table.cluster_key_id(),
                 )?;
             }
             None => {
diff --git a/src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs b/src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs
index 3cf8f498a9e2f..4146b6840c7b2 100644
--- a/src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs
+++ b/src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs
@@ -197,12 +197,12 @@ impl TableMutationAggregator {
         kind: MutationKind,
         table_meta_timestamps: TableMetaTimestamps,
     ) -> Self {
-        let fill_missing_cluster_stats = table.resolve_physical_cluster_keys().is_some();
+        let fill_missing_cluster_stats = table.resolve_cluster_keys().is_some();
 
         let virtual_schema = table.table_info.meta.virtual_schema.clone();
         let cluster_key_exprs = if fill_missing_cluster_stats {
             table
-                .resolve_physical_cluster_keys()
+                .resolve_cluster_keys()
                 .map(|cluster_keys| {
                     parse_cluster_keys(ctx.clone(), Arc::new(table.clone()), cluster_keys)
                 })
@@ -216,7 +216,7 @@ impl TableMutationAggregator {
             dal: table.get_operator(),
             location_gen: table.meta_location_generator().clone(),
             thresholds: table.get_block_thresholds(),
-            default_cluster_key: table.physical_cluster_key_id(),
+            default_cluster_key: table.cluster_key_id(),
             partition_key_count: table.partition_key_count(),
             cluster_key_exprs: Arc::from(cluster_key_exprs.clone()),
             schema: table.schema(),
@@ -386,9 +386,8 @@ impl TableMutationAggregator {
                 .and_then(|group| group.last())
                 .is_none_or(|previous| {
                     !same_partition(
-                        previous.0.cluster_stats.as_ref(),
-                        block.0.cluster_stats.as_ref(),
-                        self.write_segment_ctx.default_cluster_key,
+                        previous.0.partition_stats.as_ref(),
+                        block.0.partition_stats.as_ref(),
                         self.write_segment_ctx.partition_key_count,
                     )
                 })
@@ -885,7 +884,7 @@ impl WriteSegmentCtx {
             .location_gen
             .gen_segment_info_location(self.table_meta_timestamps, false);
         let mut new_summary =
-            reduce_block_metas(&blocks, self.thresholds, self.default_cluster_key);
+            reduce_block_metas(&blocks, self.thresholds, self.default_cluster_key)?;
         if force_all_blocks_perfect {
             // To fix issue #13217.
             if new_summary.block_count > new_summary.perfect_block_count {
diff --git a/src/query/storages/fuse/src/operations/common/processors/transform_partition_by.rs b/src/query/storages/fuse/src/operations/common/processors/transform_partition_by.rs
index 29ef8a9a60f03..f670e88f69261 100644
--- a/src/query/storages/fuse/src/operations/common/processors/transform_partition_by.rs
+++ b/src/query/storages/fuse/src/operations/common/processors/transform_partition_by.rs
@@ -28,11 +28,10 @@ use databend_common_pipeline_transforms::AccumulatingTransform;
 
 use crate::operations::mutation::SerializeDataMeta;
 
-/// Splits a block that is already sorted by the physical Fuse key into blocks
-/// whose partition-key prefix is constant.
+/// Splits a block whenever adjacent partition-key values differ.
 ///
-/// This is a per-block boundary invariant; fragments of the same partition from
-/// different input blocks are not merged here.
+/// It does not reorder rows or merge non-contiguous fragments of the same
+/// partition. Higher-level compaction is responsible for normalizing them.
 pub struct TransformPartitionBy {
     partition_key_indices: Arc<[usize]>,
     rewrite_replaced_block: bool,
@@ -56,15 +55,13 @@ impl TransformPartitionBy {
     }
 }
 
-fn partition_boundaries(block: &DataBlock, indices: &[usize]) -> Vec<usize> {
+fn partition_boundaries(block: &DataBlock, indices: &[usize]) -> impl Iterator<Item = usize> {
     let rows = block.num_rows();
-    if rows <= 1 || indices.is_empty() {
-        return Vec::new();
-    }
-
     let mut changed = vec![false; rows];
-    for index in indices {
-        mark_column_boundaries(block.get_by_offset(*index), &mut changed);
+    if rows > 1 {
+        for index in indices {
+            mark_column_boundaries(block.get_by_offset(*index), &mut changed);
+        }
     }
 
     changed
@@ -72,7 +69,6 @@ fn partition_boundaries(block: &DataBlock, indices: &[usize]) -> Vec<usize> {
         .enumerate()
         .skip(1)
         .filter_map(|(row, changed)| changed.then_some(row))
-        .collect()
 }
 
 fn mark_column_boundaries(entry: &BlockEntry, changed: &mut [bool]) {
@@ -162,10 +158,9 @@ impl AccumulatingTransform for TransformPartitionBy {
         } else {
             None
         };
-        let boundaries = partition_boundaries(&block, &self.partition_key_indices);
-        let mut blocks = Vec::with_capacity(boundaries.len() + 1);
+        let mut blocks = Vec::new();
         let mut start = 0;
-        for end in boundaries {
+        for end in partition_boundaries(&block, &self.partition_key_indices) {
             blocks.push(block.slice(start..end));
             start = end;
         }
@@ -211,7 +206,10 @@ mod tests {
             Int32Type::from_data(vec![0, 0, 1, 1, 1]),
             StringType::from_data(vec!["a", "a", "a", "b", "b"]),
         ]);
-        assert_eq!(partition_boundaries(&block, &[0, 1]), vec![2, 3]);
+        assert_eq!(
+            partition_boundaries(&block, &[0, 1]).collect::<Vec<_>>(),
+            vec![2, 3]
+        );
 
         let block = DataBlock::new_from_columns(vec![Int32Type::from_opt_data(vec![
             None,
@@ -220,22 +218,25 @@ mod tests {
             Some(1),
             Some(2),
         ])]);
-        assert_eq!(partition_boundaries(&block, &[0]), vec![2, 4]);
+        assert_eq!(
+            partition_boundaries(&block, &[0]).collect::<Vec<_>>(),
+            vec![2, 4]
+        );
     }
 
     #[test]
-    fn test_split_sorted_block_by_partition_prefix() -> Result<()> {
+    fn test_split_block_without_regrouping() -> Result<()> {
         let block = DataBlock::new_from_columns(vec![
-            Int32Type::from_data(vec![0, 0, 1, 1, 2]),
-            Int32Type::from_data(vec![1, 2, 1, 2, 1]),
+            Int32Type::from_data(vec![0, 1, 0, 1, 2]),
+            Int32Type::from_data(vec![1, 1, 2, 2, 3]),
         ]);
         let mut transform = TransformPartitionBy::new(Arc::from([0]));
         let blocks = transform.transform(block)?;
 
-        assert_eq!(blocks.len(), 3);
+        assert_eq!(blocks.len(), 5);
         assert_eq!(
             blocks.iter().map(DataBlock::num_rows).collect::<Vec<_>>(),
-            vec![2, 2, 1]
+            vec![1, 1, 1, 1, 1]
         );
         Ok(())
     }
diff --git a/src/query/storages/fuse/src/operations/common/processors/transform_serialize_segment.rs b/src/query/storages/fuse/src/operations/common/processors/transform_serialize_segment.rs
index d924ca9adc488..de2f600b3b31c 100644
--- a/src/query/storages/fuse/src/operations/common/processors/transform_serialize_segment.rs
+++ b/src/query/storages/fuse/src/operations/common/processors/transform_serialize_segment.rs
@@ -112,7 +112,7 @@ impl<B: SegmentBuilder> TransformSerializeSegment<B> {
         let virtual_column_accumulator =
             VirtualColumnAccumulator::try_create(&table_meta.schema, &table_meta.virtual_schema);
 
-        let default_cluster_key_id = table.physical_cluster_key_id();
+        let default_cluster_key_id = table.cluster_key_id();
 
         let block_top_n_template =
             table
@@ -264,13 +264,17 @@ impl<B: SegmentBuilder> Processor for TransformSerializeSegment<B> {
             };
 
             let next_partition = partition_values(
-                extended_block_meta.block_meta.cluster_stats.as_ref(),
-                self.default_cluster_key_id,
+                extended_block_meta.block_meta.partition_stats.as_ref(),
                 self.partition_key_count,
             );
+            if self.partition_key_count != 0 && next_partition.is_none() {
+                return Err(ErrorCode::Internal(
+                    "partitioned block is missing valid partition statistics",
+                ));
+            }
             if self.segment_builder.block_count() != 0
                 && self.partition_key_count != 0
-                && (next_partition.is_none() || self.current_partition.as_deref() != next_partition)
+                && self.current_partition.as_deref() != next_partition
             {
                 self.pending_block = Some(extended_block_meta);
                 self.state = State::GenerateSegment;
diff --git a/src/query/storages/fuse/src/operations/compact.rs b/src/query/storages/fuse/src/operations/compact.rs
index 67f4d3d668873..bbedbd02ba3a3 100644
--- a/src/query/storages/fuse/src/operations/compact.rs
+++ b/src/query/storages/fuse/src/operations/compact.rs
@@ -75,7 +75,7 @@ impl FuseTable {
             compact_options,
             self.meta_location_generator().clone(),
             self.operator.clone(),
-            self.physical_cluster_key_id(),
+            self.cluster_key_id(),
             table_meta_timestamps,
         )?;
         segment_compactor.partition_key_count = self.partition_key_count();
@@ -166,7 +166,7 @@ impl FuseTable {
             thresholds,
             compact_options,
             self.operator.clone(),
-            self.physical_cluster_key_id(),
+            self.cluster_key_id(),
         );
         mutator.partition_key_count = self.partition_key_count();
 
diff --git a/src/query/storages/fuse/src/operations/mutation/mutator/block_compact_mutator.rs b/src/query/storages/fuse/src/operations/mutation/mutator/block_compact_mutator.rs
index 21ff66c483de4..d008196d743b9 100644
--- a/src/query/storages/fuse/src/operations/mutation/mutator/block_compact_mutator.rs
+++ b/src/query/storages/fuse/src/operations/mutation/mutator/block_compact_mutator.rs
@@ -397,9 +397,8 @@ impl SegmentCompactChecker {
         let mut output = Vec::new();
         if let Some((_, previous)) = self.segments.last()
             && !same_partition(
-                previous.summary.cluster_stats.as_ref(),
-                segment.summary.cluster_stats.as_ref(),
-                self.cluster_key_id,
+                previous.summary.partition_stats.as_ref(),
+                segment.summary.partition_stats.as_ref(),
                 self.partition_key_count,
             )
         {
@@ -679,8 +678,7 @@ impl CompactTaskBuilder {
         let mut seen_block = false;
         for (block_meta, hlls) in blocks.iter() {
             let current_partition = partition_values(
-                block_meta.cluster_stats.as_ref(),
-                self.cluster_key_id,
+                block_meta.partition_stats.as_ref(),
                 self.partition_key_count,
             );
             if seen_block
diff --git a/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs b/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs
index b0bcc8d1a535d..bd441ab42c35e 100644
--- a/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs
+++ b/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs
@@ -301,6 +301,7 @@ struct VectorReclusterSegment {
     segment: SelectedReclusterSegment,
     block_metas: Vec<Arc<BlockMeta>>,
     stats: ClusterStatistics,
+    partition: Option<Vec<Scalar>>,
 }
 
 #[derive(Clone)]
@@ -328,7 +329,7 @@ impl ReclusterMutator {
         mode: ReclusterMode,
     ) -> Result<Self> {
         let schema = table.schema_with_stream();
-        let cluster_key_id = table.physical_cluster_key_id().unwrap();
+        let cluster_key_id = table.cluster_key_id().unwrap();
         let block_thresholds = table.get_block_thresholds();
 
         let depth_threshold = table
@@ -353,7 +354,7 @@ impl ReclusterMutator {
         }
 
         // safe to unwrap
-        let cluster_keys = table.resolve_physical_cluster_keys().unwrap();
+        let cluster_keys = table.resolve_cluster_keys().unwrap();
         let full_cluster_key_exprs =
             parse_cluster_keys(ctx.clone(), Arc::new(table.clone()), cluster_keys)?;
         let vector_cluster_info = vector_cluster_info_from_exprs(table, &full_cluster_key_exprs)?;
@@ -397,6 +398,7 @@ impl ReclusterMutator {
         depth_threshold: f64,
         block_thresholds: BlockThresholds,
         cluster_key_id: u32,
+        partition_key_count: usize,
         max_tasks: usize,
         mode: ReclusterMode,
         vector_cluster_info: Option<VectorClusterInfo>,
@@ -425,7 +427,7 @@ impl ReclusterMutator {
             depth_threshold,
             block_thresholds,
             cluster_key_id,
-            partition_key_count: 0,
+            partition_key_count,
             max_tasks,
             memory_threshold,
             prepared_cluster_key_exprs,
@@ -549,7 +551,8 @@ impl ReclusterMutator {
         blocks: &[&ReclusterBlock],
         task_budget: usize,
     ) -> Result<Vec<ReclusterTaskCandidate>> {
-        let mut blocks_map: BTreeMap<(ReclusterGroup, Vec<Scalar>), Vec<usize>> = BTreeMap::new();
+        let mut blocks_map: BTreeMap<(ReclusterGroup, Option<Vec<Scalar>>), Vec<usize>> =
+            BTreeMap::new();
         for (idx, block) in blocks.iter().enumerate() {
             let level = block.stats().level;
             if level < 0 {
@@ -559,16 +562,16 @@ impl ReclusterMutator {
                 // Terminal-level blocks are excluded from further rewrite tasks.
                 continue;
             }
-            let partition = if self.partition_key_count == 0 {
-                Vec::new()
-            } else if let Some(partition) = partition_values(
-                Some(block.stats()),
-                Some(self.cluster_key_id),
-                self.partition_key_count,
-            ) {
-                partition.to_vec()
+            let partition = if self.partition_key_count > 0 {
+                let Some(values) = partition_values(
+                    block.meta.partition_stats.as_ref(),
+                    self.partition_key_count,
+                ) else {
+                    continue;
+                };
+                Some(values.to_vec())
             } else {
-                continue;
+                None
             };
             blocks_map
                 .entry((ReclusterGroup::assign(level, self.mode), partition))
@@ -1142,11 +1145,23 @@ impl ReclusterMutator {
 
         let mut total_blocks = 0;
         let mut segments = vec![None; compact_segments.len()];
-        let mut segment_points: HashMap<Vec<Scalar>, (Vec<usize>, Vec<usize>)> = HashMap::new();
+        // Partition is a selection boundary, not part of the cluster range.
+        let mut segment_points_by_partition = BTreeMap::new();
 
         // Phase 1: collect segment ranges for the sweep-line selection. Large
         // unclustered segments are skipped because rewriting them is not useful.
         for (i, (loc, compact_segment)) in compact_segments.iter().enumerate() {
+            let partition = if self.partition_key_count > 0 {
+                let Some(values) = partition_values(
+                    compact_segment.summary.partition_stats.as_ref(),
+                    self.partition_key_count,
+                ) else {
+                    continue;
+                };
+                Some(values.to_vec())
+            } else {
+                None
+            };
             let stats = self.build_cluster_stats_for_recluster(
                 compact_segment.summary.cluster_stats.as_ref(),
                 &compact_segment.summary.col_stats,
@@ -1158,6 +1173,8 @@ impl ReclusterMutator {
             }
 
             total_blocks += compact_segment.summary.block_count as usize;
+            let segment_points: &mut HashMap<Vec<Scalar>, (Vec<usize>, Vec<usize>)> =
+                segment_points_by_partition.entry(partition).or_default();
             segment_points
                 .entry(stats.min().clone())
                 .and_modify(|v| v.0.push(i))
@@ -1179,78 +1196,55 @@ impl ReclusterMutator {
         // one window (at its start point), so windows never share a segment and
         // tasks never read the same block twice. A window closes at `window_len`;
         // `prev_window` holds the last closed one so a small tail can fold into it.
-        let mut unfinished_intervals = BTreeMap::new();
-        let mut prev_window: Option<(IndexSet<usize>, usize)> = None;
-        let mut current_window: IndexSet<usize> = IndexSet::new();
-        let mut current_window_max_depth = 0usize;
-        let (keys, values): (Vec<_>, Vec<_>) = segment_points.into_iter().unzip();
-        let sorted_indices = compare_scalars(&keys, &self.cluster_key_types)?;
-        let mut previous_partition: Option<Vec<Scalar>> = None;
-
-        for idx in sorted_indices {
-            let point = &keys[idx as usize];
-            let point_partition = (self.partition_key_count != 0
-                && point.len() >= self.partition_key_count)
-                .then(|| point[..self.partition_key_count].to_vec());
-            if previous_partition.is_some()
-                && (point_partition.is_none() || previous_partition != point_partition)
-            {
-                // A sweep window is never allowed to span physical partitions. In
-                // particular, this prevents the repack-only path from combining
-                // otherwise disjoint partition segments.
-                if let Some((segs, depth)) = prev_window.take() {
-                    windows.push((segs, depth));
-                }
-                if !current_window.is_empty() {
-                    windows.push((
+        for segment_points in segment_points_by_partition.into_values() {
+            let mut unfinished_intervals = BTreeMap::new();
+            let mut prev_window: Option<(IndexSet<usize>, usize)> = None;
+            let mut current_window: IndexSet<usize> = IndexSet::new();
+            let mut current_window_max_depth = 0usize;
+            let (keys, values): (Vec<_>, Vec<_>) = segment_points.into_iter().unzip();
+            let sorted_indices = compare_scalars(&keys, &self.cluster_key_types)?;
+
+            for idx in sorted_indices {
+                let start = &values[idx as usize].0;
+                let end = &values[idx as usize].1;
+                let point_depth = Self::calc_point_depth(unfinished_intervals.len(), start, end);
+
+                // A window is just a contiguous run of segments, so partitioning the
+                // run keeps windows segment-disjoint without any extra bookkeeping.
+                // Depth only contributes to the window score.
+                current_window_max_depth = current_window_max_depth.max(point_depth);
+                current_window.extend(start.iter().copied());
+
+                if current_window.len() >= window_len {
+                    // Emit the previously closed window and rotate the current
+                    // window into `prev_window`.
+                    if let Some((segs, depth)) = prev_window.take() {
+                        windows.push((segs, depth));
+                    }
+                    prev_window = Some((
                         std::mem::take(&mut current_window),
                         current_window_max_depth,
                     ));
+                    current_window_max_depth = 0;
                 }
-                current_window_max_depth = 0;
-                unfinished_intervals.clear();
-            }
-            previous_partition = point_partition;
-
-            let start = &values[idx as usize].0;
-            let end = &values[idx as usize].1;
-            let point_depth = Self::calc_point_depth(unfinished_intervals.len(), start, end);
-
-            // A window is just a contiguous run of segments, so partitioning the
-            // run keeps windows segment-disjoint without any extra bookkeeping.
-            // Depth only contributes to the window score.
-            current_window_max_depth = current_window_max_depth.max(point_depth);
-            current_window.extend(start.iter().copied());
-
-            if current_window.len() >= window_len {
-                // Emit the previously closed window and rotate the current
-                // window into `prev_window`.
-                if let Some((segs, depth)) = prev_window.take() {
-                    windows.push((segs, depth));
-                }
-                prev_window = Some((
-                    std::mem::take(&mut current_window),
-                    current_window_max_depth,
-                ));
-                current_window_max_depth = 0;
-            }
 
-            start.iter().for_each(|&idx| {
-                unfinished_intervals.insert(idx, point_depth);
-            });
-            end.iter().for_each(|idx| {
-                unfinished_intervals.remove(idx);
-            });
-        }
+                start.iter().for_each(|&idx| {
+                    unfinished_intervals.insert(idx, point_depth);
+                });
+                end.iter().for_each(|idx| {
+                    unfinished_intervals.remove(idx);
+                });
+            }
 
-        // Fold the trailing window into the last closed one to avoid a tiny
-        // fragment; this may push it past `window_len` (an acceptable soft
-        // overshoot under `LIMIT`).
-        if let Some((mut prev_segs, prev_depth)) = prev_window.take() {
-            prev_segs.extend(current_window);
-            windows.push((prev_segs, prev_depth.max(current_window_max_depth)));
-        } else if !current_window.is_empty() {
-            windows.push((current_window, current_window_max_depth));
+            // Fold the trailing window into the last closed one to avoid a tiny
+            // fragment; this may push it past `window_len` (an acceptable soft
+            // overshoot under `LIMIT`).
+            if let Some((mut prev_segs, prev_depth)) = prev_window.take() {
+                prev_segs.extend(current_window);
+                windows.push((prev_segs, prev_depth.max(current_window_max_depth)));
+            } else if !current_window.is_empty() {
+                windows.push((current_window, current_window_max_depth));
+            }
         }
 
         // Try the deepest windows first; for equal depth, prefer the larger
@@ -1304,6 +1298,17 @@ impl ReclusterMutator {
         let mut small_segments = IndexSet::new();
 
         for (loc, compact_segment) in compact_segments {
+            let partition = if self.partition_key_count > 0 {
+                let Some(values) = partition_values(
+                    compact_segment.summary.partition_stats.as_ref(),
+                    self.partition_key_count,
+                ) else {
+                    continue;
+                };
+                Some(values.to_vec())
+            } else {
+                None
+            };
             let stats = self.build_cluster_stats_for_recluster(
                 compact_segment.summary.cluster_stats.as_ref(),
                 &compact_segment.summary.col_stats,
@@ -1328,6 +1333,7 @@ impl ReclusterMutator {
                 segment,
                 block_metas,
                 stats,
+                partition,
             });
         }
 
@@ -1377,21 +1383,23 @@ impl ReclusterMutator {
             );
         }
 
-        let mut fallback_indices = Vec::new();
-        for idx in 0..vector_segments.len() {
+        let mut fallback_indices = BTreeMap::<Option<Vec<Scalar>>, Vec<usize>>::new();
+        for (idx, segment) in vector_segments.iter().enumerate() {
             if !covered_segments.contains(&idx) || small_segments.contains(&idx) {
-                fallback_indices.push(idx);
+                fallback_indices
+                    .entry(segment.partition.clone())
+                    .or_default()
+                    .push(idx);
             }
         }
 
-        for selected_indices in fallback_indices
-            .chunks(window_len)
-            .map(|chunk| chunk.to_vec())
-        {
-            let mut window_key = selected_indices.clone();
-            window_key.sort_unstable();
-            if seen_windows.insert(window_key) {
-                windows.push((selected_indices, 0));
+        for indices in fallback_indices.into_values() {
+            for selected_indices in indices.chunks(window_len).map(|chunk| chunk.to_vec()) {
+                let mut window_key = selected_indices.clone();
+                window_key.sort_unstable();
+                if seen_windows.insert(window_key) {
+                    windows.push((selected_indices, 0));
+                }
             }
         }
 
@@ -1436,6 +1444,8 @@ impl ReclusterMutator {
                     segment,
                     block_metas,
                     stats,
+                    // Scalar windows have already been isolated by partition.
+                    partition: None,
                 });
             }
 
@@ -1490,11 +1500,12 @@ impl ReclusterMutator {
 
         for left in 0..vector_segments.len() {
             for right in left + 1..vector_segments.len() {
-                if require_scalar_overlap
-                    && !self.scalar_cluster_stats_overlap(
-                        &vector_segments[left].stats,
-                        &vector_segments[right].stats,
-                    )
+                if vector_segments[left].partition != vector_segments[right].partition
+                    || require_scalar_overlap
+                        && !self.scalar_cluster_stats_overlap(
+                            &vector_segments[left].stats,
+                            &vector_segments[right].stats,
+                        )
                 {
                     continue;
                 }
diff --git a/src/query/storages/fuse/src/operations/mutation/mutator/segment_compact_mutator.rs b/src/query/storages/fuse/src/operations/mutation/mutator/segment_compact_mutator.rs
index ce2fc38754cab..709e525faf230 100644
--- a/src/query/storages/fuse/src/operations/mutation/mutator/segment_compact_mutator.rs
+++ b/src/query/storages/fuse/src/operations/mutation/mutator/segment_compact_mutator.rs
@@ -330,9 +330,8 @@ impl<'a> SegmentCompactor<'a> {
 
         if let Some((_, previous, _)) = self.fragmented_segments.last()
             && !same_partition(
-                previous.summary.cluster_stats.as_ref(),
-                segment_info.summary.cluster_stats.as_ref(),
-                self.default_cluster_key_id,
+                previous.summary.partition_stats.as_ref(),
+                segment_info.summary.partition_stats.as_ref(),
                 self.partition_key_count,
             )
         {
diff --git a/src/query/storages/fuse/src/operations/recluster.rs b/src/query/storages/fuse/src/operations/recluster.rs
index 73ad8473edb17..6af0ac46d7dee 100644
--- a/src/query/storages/fuse/src/operations/recluster.rs
+++ b/src/query/storages/fuse/src/operations/recluster.rs
@@ -65,7 +65,7 @@ impl FuseTable {
 
         ctx.set_status_info("[FUSE-RECLUSTER] Starting recluster operation");
 
-        if self.physical_cluster_key_id().is_none() {
+        if self.cluster_key_id().is_none() {
             return Ok(None);
         }
 
diff --git a/src/query/storages/fuse/src/pruning/partition_pruner.rs b/src/query/storages/fuse/src/pruning/partition_pruner.rs
index dd5e1effd9c75..74f7c98f4cb04 100644
--- a/src/query/storages/fuse/src/pruning/partition_pruner.rs
+++ b/src/query/storages/fuse/src/pruning/partition_pruner.rs
@@ -32,18 +32,16 @@ use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_sql::plans::ComparisonOp;
 use databend_common_statistics::HistogramBounds;
 use databend_common_statistics::HistogramRangeBounds;
-use databend_storages_common_table_meta::meta::ClusterStatistics;
+use databend_storages_common_table_meta::meta::PartitionStatistics;
 
 use crate::statistics::partition_values;
 
 #[derive(Clone)]
 pub struct PartitionPruningInfo {
-    pub cluster_key_id: u32,
     pub partition_keys: Vec<RemoteExpr<String>>,
 }
 
 pub struct PartitionPruner {
-    cluster_key_id: u32,
     partition_keys: Vec<Expr<String>>,
     filter: Expr<String>,
     func_ctx: FunctionContext,
@@ -75,17 +73,14 @@ impl PartitionPruner {
             .collect();
 
         Some(Self {
-            cluster_key_id: info.cluster_key_id,
             partition_keys,
             filter,
             func_ctx,
         })
     }
 
-    pub fn should_keep(&self, stats: Option<&ClusterStatistics>) -> bool {
-        let Some(values) =
-            partition_values(stats, Some(self.cluster_key_id), self.partition_keys.len())
-        else {
+    pub fn should_keep(&self, stats: Option<&PartitionStatistics>) -> bool {
+        let Some(values) = partition_values(stats, self.partition_keys.len()) else {
             return true;
         };
 
@@ -534,13 +529,12 @@ mod tests {
         }]);
 
         let pruner = PartitionPruner {
-            cluster_key_id: 0,
             partition_keys: vec![partition_expr],
             filter,
             func_ctx: FunctionContext::default(),
         };
 
-        // No cluster stats at all — must keep the segment.
+        // No partition stats at all — must keep the segment.
         assert!(pruner.should_keep(None));
     }
 }
diff --git a/src/query/storages/fuse/src/pruning/segment_pruner.rs b/src/query/storages/fuse/src/pruning/segment_pruner.rs
index 9897b2f59f728..f7d2614257f07 100644
--- a/src/query/storages/fuse/src/pruning/segment_pruner.rs
+++ b/src/query/storages/fuse/src/pruning/segment_pruner.rs
@@ -93,7 +93,9 @@ impl SegmentPruner {
                 self.pruning_ctx
                     .partition_pruner
                     .as_ref()
-                    .is_none_or(|pruner| pruner.should_keep(info.summary().cluster_stats.as_ref()))
+                    .is_none_or(|pruner| {
+                        pruner.should_keep(info.summary().partition_stats.as_ref())
+                    })
                     && range_pruner.should_keep(&range_input, None)
             }) {
                 // Perf.
diff --git a/src/query/storages/fuse/src/retry/commit.rs b/src/query/storages/fuse/src/retry/commit.rs
index 46104ed46ea6c..e5c22bcd3301d 100644
--- a/src/query/storages/fuse/src/retry/commit.rs
+++ b/src/query/storages/fuse/src/retry/commit.rs
@@ -186,7 +186,7 @@ async fn try_rebuild_req(
             storage_class,
             table_info.desc.as_str(),
         )?;
-        let default_cluster_key_id = latest_table.physical_cluster_key_id();
+        let default_cluster_key_id = latest_table.cluster_key_id();
         let latest_snapshot = latest_table.read_table_snapshot().await?;
         let (update_table_meta_req, _) = req
             .update_table_metas
diff --git a/src/query/storages/fuse/src/statistics/accumulator.rs b/src/query/storages/fuse/src/statistics/accumulator.rs
index dbb586347f0ae..aeb4258b4e633 100644
--- a/src/query/storages/fuse/src/statistics/accumulator.rs
+++ b/src/query/storages/fuse/src/statistics/accumulator.rs
@@ -63,7 +63,7 @@ impl SegmentBuilder for RowOrientedSegmentBuilder {
     ) -> Result<Self::Segment> {
         let builder = std::mem::take(self);
         let mut stat =
-            super::reduce_block_metas(&builder.blocks_metas, thresholds, default_cluster_key_id);
+            super::reduce_block_metas(&builder.blocks_metas, thresholds, default_cluster_key_id)?;
         stat.additional_stats_meta = additional_stats_meta;
         Ok(SegmentInfo::new(builder.blocks_metas, stat))
     }
diff --git a/src/query/storages/fuse/src/statistics/cluster_statistics.rs b/src/query/storages/fuse/src/statistics/cluster_statistics.rs
index 74934deff3067..7ffac81383f98 100644
--- a/src/query/storages/fuse/src/statistics/cluster_statistics.rs
+++ b/src/query/storages/fuse/src/statistics/cluster_statistics.rs
@@ -16,6 +16,7 @@ use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::collections::HashMap;
 
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::BlockThresholds;
 use databend_common_expression::ColumnId;
@@ -37,6 +38,7 @@ use databend_common_meta_app::schema::TableIndexType;
 use databend_common_sql::evaluator::BlockOperator;
 use databend_storages_common_index::statistics_to_domain;
 use databend_storages_common_table_meta::meta::ClusterStatistics;
+use databend_storages_common_table_meta::meta::PartitionStatistics;
 use databend_storages_common_table_meta::meta::StatisticsOfColumns;
 use databend_storages_common_table_meta::meta::VectorDistanceType;
 use log::warn;
@@ -89,7 +91,7 @@ pub struct ClusterStatsGenerator {
     block_thresholds: BlockThresholds,
 
     pub extra_key_num: usize,
-    pub partition_key_count: usize,
+    pub partition_key_index: Vec<usize>,
     pub cluster_key_index: Vec<usize>,
     pub operators: Vec<BlockOperator>,
     pub vector_operator: Option<VectorClusterOperator>,
@@ -113,7 +115,7 @@ impl ClusterStatsGenerator {
         Self {
             cluster_key_id,
             cluster_key_index,
-            partition_key_count: 0,
+            partition_key_index: Vec::new(),
             extra_key_num,
             level,
             block_thresholds,
@@ -156,6 +158,63 @@ impl ClusterStatsGenerator {
         Ok((cluster_stats, data_block))
     }
 
+    pub fn extract_partition_stats(
+        &self,
+        data_block: &DataBlock,
+    ) -> Result<Option<PartitionStatistics>> {
+        if self.partition_key_index.is_empty() {
+            return Ok(None);
+        }
+        if data_block.is_empty() {
+            return Err(ErrorCode::Internal(
+                "cannot extract partition statistics from an empty block",
+            ));
+        }
+
+        // Append blocks already contain evaluated key expressions. Mutation paths may
+        // serialize source columns directly, so evaluate the key operators on demand.
+        let evaluated_block = if self
+            .partition_key_index
+            .iter()
+            .any(|index| *index >= data_block.num_columns())
+        {
+            Some(
+                self.operators
+                    .iter()
+                    .try_fold(data_block.clone(), |block, operator| {
+                        operator.execute(&self.func_ctx, block)
+                    })?,
+            )
+        } else {
+            None
+        };
+        let data_block = evaluated_block.as_ref().unwrap_or(data_block);
+        if self
+            .partition_key_index
+            .iter()
+            .any(|index| *index >= data_block.num_columns())
+        {
+            return Err(ErrorCode::Internal(
+                "partition key columns are missing before block serialization",
+            ));
+        }
+
+        let mut values = Vec::with_capacity(self.partition_key_index.len());
+        for index in &self.partition_key_index {
+            let entry = data_block.get_by_offset(*index);
+            let value = entry
+                .index(0)
+                .ok_or_else(|| ErrorCode::Internal("partition key is missing from block"))?;
+            if (1..data_block.num_rows()).any(|row| entry.index(row).as_ref() != Some(&value)) {
+                return Err(ErrorCode::Internal(
+                    "serialized block contains more than one partition",
+                ));
+            }
+            values.push(value.to_owned());
+        }
+        Ok(Some(PartitionStatistics::new(values)))
+    }
+
     // This can be used in deletion, for an existing block.
     pub fn gen_with_origin_stats(
         &self,
@@ -296,40 +355,26 @@ pub fn sort_by_cluster_stats(
     }
 }
 
-/// Returns the exact partition prefix stored in physical cluster statistics.
-/// A partitioned block/segment is valid only when every prefix dimension has
-/// identical min and max values.
 pub(crate) fn partition_values(
-    stats: Option<&ClusterStatistics>,
-    cluster_key_id: Option<u32>,
+    stats: Option<&PartitionStatistics>,
     partition_key_count: usize,
 ) -> Option<&[Scalar]> {
-    if partition_key_count == 0 {
-        return None;
-    }
-    let stats = stats?;
-    if Some(stats.cluster_key_id) != cluster_key_id
-        || stats.min.len() < partition_key_count
-        || stats.max.len() < partition_key_count
-        || stats.min[..partition_key_count] != stats.max[..partition_key_count]
-    {
-        return None;
-    }
-    Some(&stats.min[..partition_key_count])
+    stats
+        .map(|stats| stats.values.as_slice())
+        .filter(|values| values.len() == partition_key_count)
 }
 
 pub(crate) fn same_partition(
-    left: Option<&ClusterStatistics>,
-    right: Option<&ClusterStatistics>,
-    cluster_key_id: Option<u32>,
+    left: Option<&PartitionStatistics>,
+    right: Option<&PartitionStatistics>,
     partition_key_count: usize,
 ) -> bool {
     if partition_key_count == 0 {
         return true;
     }
     match (
-        partition_values(left, cluster_key_id, partition_key_count),
-        partition_values(right, cluster_key_id, partition_key_count),
+        partition_values(left, partition_key_count),
+        partition_values(right, partition_key_count),
     ) {
         (Some(left), Some(right)) => left == right,
         _ => false,
@@ -560,8 +605,12 @@ pub(crate) fn get_min_max_stats(
 #[cfg(test)]
 mod tests {
     use databend_common_expression::ColumnRef;
+    use databend_common_expression::FromData;
+    use databend_common_expression::RawExpr;
     use databend_common_expression::TableDataType;
     use databend_common_expression::TableField;
+    use databend_common_expression::type_check::check;
+    use databend_common_expression::types::Int32Type;
     use databend_common_expression::types::NumberDataType;
     use databend_common_expression::types::number::NumberScalar;
     use databend_storages_common_table_meta::meta::ColumnStatistics;
@@ -648,57 +697,91 @@ mod tests {
 
     #[test]
     fn test_partition_values_returns_none_for_missing_stats() {
-        // When segment has no cluster statistics, pruner must keep it conservatively.
-        assert_eq!(partition_values(None, Some(0), 1), None);
-    }
-
-    #[test]
-    fn test_partition_values_returns_none_for_mismatched_cluster_key_id() {
-        let stats = ClusterStatistics::new(0, vec![int32_scalar(1)], vec![int32_scalar(1)], 0);
-        // Segment was written with a different cluster key id; treat as unpartitioned.
-        assert_eq!(partition_values(Some(&stats), Some(1), 1), None);
+        assert_eq!(partition_values(None, 1), None);
     }
 
     #[test]
-    fn test_partition_values_returns_none_when_prefix_min_max_differ() {
-        // A block that spans two partition values is not a valid single-partition block.
-        let stats = ClusterStatistics::new(0, vec![int32_scalar(1)], vec![int32_scalar(2)], 0);
-        assert_eq!(partition_values(Some(&stats), Some(0), 1), None);
+    fn test_partition_values_returns_exact_values() {
+        let stats = PartitionStatistics::new(vec![int32_scalar(3)]);
+        let values = partition_values(Some(&stats), 1).unwrap();
+        assert_eq!(values, &[int32_scalar(3)]);
+        assert_eq!(partition_values(Some(&stats), 2), None);
     }
 
     #[test]
-    fn test_partition_values_returns_prefix_when_valid() {
-        let stats = ClusterStatistics::new(
-            0,
-            vec![int32_scalar(3), int32_scalar(10)],
-            vec![int32_scalar(3), int32_scalar(20)],
+    fn test_partition_and_cluster_statistics_are_independent() -> Result<()> {
+        let block = DataBlock::new_from_columns(vec![
+            Int32Type::from_data(vec![6, 6]),
+            Int32Type::from_data(vec![1, 2]),
+        ]);
+        let partition_expr = check(
+            &RawExpr::FunctionCall {
+                span: None,
+                name: "plus".to_string(),
+                params: vec![],
+                args: vec![
+                    RawExpr::ColumnRef {
+                        span: None,
+                        id: 0,
+                        data_type: DataType::Number(NumberDataType::Int32),
+                        display_name: "p".to_string(),
+                    },
+                    RawExpr::Constant {
+                        span: None,
+                        scalar: int32_scalar(1),
+                        data_type: None,
+                    },
+                ],
+            },
+            &BUILTIN_FUNCTIONS,
+        )?;
+        let mut generator = ClusterStatsGenerator::new(
+            3,
+            vec![1],
+            1,
             0,
+            BlockThresholds::default(),
+            vec![BlockOperator::Map {
+                exprs: vec![partition_expr],
+                projections: None,
+            }],
+            None,
+            vec![],
+            FunctionContext::default(),
         );
-        // Only the partition prefix (first dimension) is returned.
-        let values = partition_values(Some(&stats), Some(0), 1).unwrap();
-        assert_eq!(values, &[int32_scalar(3)]);
+        generator.partition_key_index = vec![2];
+
+        let partition_stats = generator.extract_partition_stats(&block)?.unwrap();
+        let cluster_stats = generator.clusters_statistics(&block, 0)?.unwrap();
+
+        assert_eq!(partition_stats.values, vec![Scalar::Number(
+            NumberScalar::Int64(7)
+        )]);
+        assert_eq!(cluster_stats.min, vec![int32_scalar(1)]);
+        assert_eq!(cluster_stats.max, vec![int32_scalar(2)]);
+        Ok(())
     }
 
     #[test]
     fn test_same_partition_returns_false_when_either_side_has_no_stats() {
         // Without partition metadata on either side, compact must not merge the segments.
-        let stats = ClusterStatistics::new(0, vec![int32_scalar(1)], vec![int32_scalar(1)], 0);
-        assert!(!same_partition(None, None, Some(0), 1));
-        assert!(!same_partition(Some(&stats), None, Some(0), 1));
-        assert!(!same_partition(None, Some(&stats), Some(0), 1));
+        let stats = PartitionStatistics::new(vec![int32_scalar(1)]);
+        assert!(!same_partition(None, None, 1));
+        assert!(!same_partition(Some(&stats), None, 1));
+        assert!(!same_partition(None, Some(&stats), 1));
     }
 
     #[test]
     fn test_same_partition_returns_true_when_partition_key_count_is_zero() {
         // Tables without PARTITION BY always treat segments as the same partition.
-        assert!(same_partition(None, None, Some(0), 0));
+        assert!(same_partition(None, None, 0));
     }
 
     #[test]
     fn test_same_partition_distinguishes_different_partition_values() {
-        let left = ClusterStatistics::new(0, vec![int32_scalar(0)], vec![int32_scalar(0)], 0);
-        let right = ClusterStatistics::new(0, vec![int32_scalar(1)], vec![int32_scalar(1)], 0);
-        assert!(!same_partition(Some(&left), Some(&right), Some(0), 1));
-        assert!(same_partition(Some(&left), Some(&left), Some(0), 1));
+        let left = PartitionStatistics::new(vec![int32_scalar(0)]);
+        let right = PartitionStatistics::new(vec![int32_scalar(1)]);
+        assert!(!same_partition(Some(&left), Some(&right), 1));
+        assert!(same_partition(Some(&left), Some(&left), 1));
     }
 }
diff --git a/src/query/storages/fuse/src/statistics/reducers.rs b/src/query/storages/fuse/src/statistics/reducers.rs
index 276386a3a48b1..388eed3d14c62 100644
--- a/src/query/storages/fuse/src/statistics/reducers.rs
+++ b/src/query/storages/fuse/src/statistics/reducers.rs
@@ -16,6 +16,7 @@ use std::borrow::Borrow;
 use std::collections::HashMap;
 use std::collections::HashSet;
 
+use databend_common_exception::Result;
 use databend_common_expression::BlockThresholds;
 use databend_common_expression::ColumnId;
 use databend_common_expression::Scalar;
@@ -23,11 +24,13 @@ use databend_common_expression::types::DataType;
 use databend_storages_common_table_meta::meta::BlockMeta;
 use databend_storages_common_table_meta::meta::ClusterStatistics;
 use databend_storages_common_table_meta::meta::ColumnStatistics;
+use databend_storages_common_table_meta::meta::PartitionStatistics;
 use databend_storages_common_table_meta::meta::SpatialStatistics;
 use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::StatisticsOfColumns;
 use databend_storages_common_table_meta::meta::StatisticsOfSpatialColumns;
 use databend_storages_common_table_meta::meta::VirtualColumnMeta;
+use databend_storages_common_table_meta::meta::validate_segment_partition_statistics;
 
 const VIRTUAL_COLUMN_JSONB_TYPE: u8 = 0;
 
@@ -321,6 +324,16 @@ pub fn reduce_cluster_statistics<T: Borrow<Option<ClusterStatistics>>>(
     Some(ClusterStatistics::new(cluster_key_id, min, max, level))
 }
 
+fn merge_partition_statistics(
+    left: Option<&PartitionStatistics>,
+    right: Option<&PartitionStatistics>,
+) -> Option<PartitionStatistics> {
+    match (left, right) {
+        (Some(left), Some(right)) if left == right => Some(left.clone()),
+        _ => None,
+    }
+}
+
 pub fn merge_statistics(
     mut l: Statistics,
     r: &Statistics,
@@ -341,6 +354,7 @@ pub fn merge_statistics_mut(
         l.virtual_col_stats = r.virtual_col_stats.clone();
         l.spatial_stats = r.spatial_stats.clone();
         l.cluster_stats = r.cluster_stats.clone();
+        l.partition_stats = r.partition_stats.clone();
     } else {
         l.col_stats = reduce_block_statistics(&[&l.col_stats, &r.col_stats]);
         l.virtual_col_stats =
@@ -350,6 +364,8 @@ pub fn merge_statistics_mut(
             &[&l.cluster_stats, &r.cluster_stats],
             default_cluster_key_id,
         );
+        l.partition_stats =
+            merge_partition_statistics(l.partition_stats.as_ref(), r.partition_stats.as_ref());
     }
 
     l.row_count += r.row_count;
@@ -390,6 +406,8 @@ pub fn deduct_statistics(l: &Statistics, r: &Statistics) -> Statistics {
 
 // Deduct statistics, only be used for calculate snapshot summary.
 pub fn deduct_statistics_mut(l: &mut Statistics, r: &Statistics) {
+    // Exact partition equality cannot be reconstructed after subtraction.
+    l.partition_stats = None;
     l.row_count -= r.row_count;
     l.block_count -= r.block_count;
     l.perfect_block_count -= r.perfect_block_count;
@@ -440,7 +458,7 @@ pub fn reduce_block_metas<T: Borrow<BlockMeta>>(
     block_metas: &[T],
     thresholds: BlockThresholds,
     default_cluster_key_id: Option<u32>,
-) -> Statistics {
+) -> Result<Statistics> {
     let mut row_count: u64 = 0;
     let mut block_count: u64 = 0;
     let mut uncompressed_byte_size: u64 = 0;
@@ -458,6 +476,7 @@ pub fn reduce_block_metas<T: Borrow<BlockMeta>>(
     let len = block_metas.len();
     let mut col_stats = Vec::with_capacity(len);
     let mut cluster_stats = Vec::with_capacity(len);
+    let mut partition_stats = Vec::with_capacity(len);
     let mut virtual_col_stats = Vec::with_capacity(len);
     let mut spatial_col_stats = Vec::with_capacity(len);
 
@@ -502,12 +521,16 @@ pub fn reduce_block_metas<T: Borrow<BlockMeta>>(
         }
         col_stats.push(&b.col_stats);
         cluster_stats.push(&b.cluster_stats);
+        partition_stats.push(&b.partition_stats);
         spatial_col_stats.push(&b.spatial_stats);
     });
 
     let merged_col_stats = reduce_block_statistics(&col_stats);
     let merged_spatial_stats = reduce_spatial_statistics(&spatial_col_stats);
     let merged_cluster_stats = reduce_cluster_statistics(&cluster_stats, default_cluster_key_id);
+    let merged_partition_stats = validate_segment_partition_statistics(
+        partition_stats.into_iter().map(|stats| stats.as_ref()),
+    )?;
     let merged_virtual_col_stats = if block_count > 0 && virtual_block_count == block_count {
         let virtual_col_stats = generate_virtual_column_statistics(&virtual_col_stats);
         Some(virtual_col_stats)
@@ -523,7 +546,7 @@ pub fn reduce_block_metas<T: Borrow<BlockMeta>>(
     let spatial_index_size = Option::from(spatial_index_size).filter(|&x| x > 0);
     let virtual_column_size = Option::from(virtual_column_size).filter(|&x| x > 0);
 
-    Statistics {
+    Ok(Statistics {
         row_count,
         block_count,
         perfect_block_count,
@@ -540,7 +563,8 @@ pub fn reduce_block_metas<T: Borrow<BlockMeta>>(
         virtual_col_stats: merged_virtual_col_stats,
         spatial_stats: merged_spatial_stats,
         cluster_stats: merged_cluster_stats,
+        partition_stats: merged_partition_stats,
         virtual_block_count: merged_virtual_block_count,
         additional_stats_meta: None,
-    }
+    })
 }
diff --git a/src/query/storages/fuse/src/table_functions/clustering_information.rs b/src/query/storages/fuse/src/table_functions/clustering_information.rs
index 31f80ccf8597a..5b9e1792e137f 100644
--- a/src/query/storages/fuse/src/table_functions/clustering_information.rs
+++ b/src/query/storages/fuse/src/table_functions/clustering_information.rs
@@ -174,7 +174,7 @@ impl<'a> ClusteringInformationImpl<'a> {
         cluster_key: &Option<String>,
     ) -> Result<ClusteringInformationResponse> {
         let mut default_cluster_key_id = None;
-        let (cluster_key, mut exprs) = match (self.table.cluster_key_str(), cluster_key) {
+        let (cluster_key, exprs) = match (self.table.cluster_key_str(), cluster_key) {
             (a, Some(b)) => {
                 let (cluster_key, exprs) =
                     analyze_cluster_keys(self.ctx.clone(), Arc::new(self.table.clone()), b)?;
@@ -183,32 +183,24 @@ impl<'a> ClusteringInformationImpl<'a> {
                     .map(|expr| expr.project_column_ref(|index| Ok(index.as_usize())))
                     .collect::<Result<Vec<_>>>()?;
                 if a.is_some() && a.unwrap() == cluster_key {
-                    default_cluster_key_id = self.table.physical_cluster_key_id();
+                    default_cluster_key_id = self.table.cluster_key_id();
                 }
                 (cluster_key, exprs)
             }
             (Some(a), None) => {
-                let cluster_keys = self.table.resolve_physical_cluster_keys().unwrap();
+                let cluster_keys = self.table.resolve_cluster_keys().unwrap();
                 let exprs = parse_cluster_keys(
                     self.ctx.clone(),
                     Arc::new(self.table.clone()),
                     cluster_keys,
                 )?;
-                default_cluster_key_id = self.table.physical_cluster_key_id();
+                default_cluster_key_id = self.table.cluster_key_id();
                 (a.to_string(), exprs)
             }
             _ => {
                 unreachable!("Unclustered table {}", self.table.table_info.desc);
             }
         };
-        if default_cluster_key_id.is_some() && self.table.partition_key_count() != 0 {
-            exprs = parse_cluster_keys(
-                self.ctx.clone(),
-                Arc::new(self.table.clone()),
-                self.table.resolve_physical_cluster_keys().unwrap(),
-            )?;
-        }
-
         let cluster_type = "linear".to_string();
 
         let snapshot = self.table.read_table_snapshot().await?;
diff --git a/src/query/storages/fuse/src/table_functions/clustering_statistics.rs b/src/query/storages/fuse/src/table_functions/clustering_statistics.rs
index 2b27b80d22f94..24ff779b2f4f6 100644
--- a/src/query/storages/fuse/src/table_functions/clustering_statistics.rs
+++ b/src/query/storages/fuse/src/table_functions/clustering_statistics.rs
@@ -95,7 +95,7 @@ impl TableMetaFunc for ClusteringStatistics {
             return Ok(DataBlock::empty_with_schema(&Self::schema().into()));
         }
 
-        let cluster_keys = tbl.resolve_physical_cluster_keys().unwrap();
+        let cluster_keys = tbl.resolve_cluster_keys().unwrap();
         let exprs = parse_cluster_keys(ctx.clone(), Arc::new(tbl.clone()), cluster_keys)?;
         let scalar_exprs = exprs
             .into_iter()
diff --git a/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test b/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test
index 005c28091092e..f5ef2724b5343 100644
--- a/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test
+++ b/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test
@@ -39,11 +39,43 @@ INSERT INTO t VALUES (0, 30, 'a0'), (1, 20, 'a1'), (2, 10, 'a2')
 statement ok
 INSERT INTO t VALUES (3, 31, 'b0'), (4, 21, 'b1'), (5, 11, 'b2')
 
+# PARTITION BY is not an implicit prefix of the user-visible cluster key.
+query T
+SELECT cluster_key FROM clustering_information('db_09_0054', 't')
+----
+(c)
+
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 't') LIMIT 1
 ----
 6 6
 
+# A partition expression may also be a cluster-key expression. Sorting by
+# (c, p % 2) produces partition runs 1, 0, 1 for this input, whereas an implicit
+# partition prefix would produce only the runs 0, 1.
+statement ok
+SET max_threads=1
+
+statement ok
+CREATE TABLE overlapping_keys(p INT, c INT) PARTITION BY (p % 2) CLUSTER BY (c, p % 2) ROW_PER_BLOCK=1000 BLOCK_PER_SEGMENT=1000
+
+statement ok
+INSERT INTO overlapping_keys VALUES (0, 30), (1, 10), (2, 20), (3, 40)
+
+query II
+SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'overlapping_keys') LIMIT 1
+----
+3 3
+
+# The three cluster-ordered runs still carry exact partition metadata.
+query T
+EXPLAIN SELECT p, c FROM overlapping_keys WHERE p % 2 = 0
+----
+<slt:ignore>partitions total: 3<slt:ignore>partitions scanned: 1<slt:ignore>range pruning: 3 to 1<slt:ignore>
+
+statement ok
+UNSET max_threads
+
 # Segment compaction may combine metadata only inside one physical partition.
 statement ok
 OPTIMIZE TABLE t COMPACT SEGMENT
@@ -63,8 +95,8 @@ SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 't') LIMIT 1
 ----
 3 3
 
-# The exact partition expression is evaluated against the partition prefix in
-# Segment ClusterStatistics before BlockMeta is read.
+# The exact partition expression is evaluated against independent segment
+# partition statistics before BlockMeta is read.
 query T
 EXPLAIN SELECT p, c, v FROM t WHERE p % 3 = 1
 ----
@@ -225,7 +257,7 @@ INSERT INTO partition_only VALUES (0, 'a'), (1, 'b'), (2, 'c'), (3, 'd')
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'partition_only') LIMIT 1
 ----
-2 2
+4 4
 
 query II rowsort
 SELECT p, length(v) FROM partition_only
@@ -241,12 +273,12 @@ CREATE TABLE replace_partitioned(id INT, p INT) PARTITION BY (p % 2) ROW_PER_BLO
 statement ok
 REPLACE INTO replace_partitioned ON(id) VALUES (1, 0), (2, 1), (3, 2), (4, 3)
 
-# REPLACE INTO is sorted and split before serialization, so a mixed input block
-# produces one block and segment per physical partition.
+# REPLACE INTO does not regroup partition-only input, so interleaved partition
+# values produce one block and segment per adjacent partition run.
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'replace_partitioned') LIMIT 1
 ----
-2 2
+4 4
 
 query II rowsort
 SELECT id, p FROM replace_partitioned
@@ -275,10 +307,11 @@ SELECT id, p FROM merge_partitioned
 statement ok
 OPTIMIZE TABLE merge_partitioned COMPACT
 
+# Compaction respects existing partition boundaries without reordering segments.
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'merge_partitioned') LIMIT 1
 ----
-2 2
+3 3
 
 # Recluster overlapping cluster-key ranges inside each partition without ever
 # selecting or rewriting blocks across the physical partition boundary.
@@ -294,7 +327,7 @@ INSERT INTO recluster_partitioned VALUES (4, 31), (5, 31), (6, 11), (7, 11)
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'recluster_partitioned') LIMIT 1
 ----
-4 4
+8 8
 
 statement ok
 ALTER TABLE recluster_partitioned RECLUSTER FINAL
@@ -350,8 +383,8 @@ WHERE column_name = 'event_time'
 ----
 0
 
-# Partition splitting is local to each sorted input block. Fragmented multi-block
-# input may therefore produce several blocks for the same physical partition.
+# Partition splitting is local to each input block and does not regroup rows.
+# The exact block count therefore depends on parallel input scheduling.
 statement ok
 SET max_threads=4
 
@@ -365,9 +398,9 @@ statement ok
 INSERT INTO fragmented_input SELECT number FROM numbers(12)
 
 query I
-SELECT block_count FROM fuse_snapshot('db_09_0054', 'fragmented_input') LIMIT 1
+SELECT if(block_count BETWEEN 6 AND 12, 1, 0) FROM fuse_snapshot('db_09_0054', 'fragmented_input') LIMIT 1
 ----
-6
+1
 
 # Like CLUSTER BY, each physical key expression may reference only one source
 # column. Composite partitioning is expressed as PARTITION BY (a, b).
@@ -416,7 +449,7 @@ INSERT INTO delete_partitioned VALUES (0, 'a'), (1, 'b'), (2, 'c'), (3, 'd')
 query II
 SELECT segment_count, block_count FROM fuse_snapshot('db_09_0054', 'delete_partitioned') LIMIT 1
 ----
-2 2
+4 4
 
 statement ok
 DELETE FROM delete_partitioned WHERE v = 'b'