From 97cf868f3101f380ac14b245617c0eedebe2896d Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 7 Aug 2026 08:27:47 -0700 Subject: [PATCH 1/2] fix(scan): preserve partition group order --- crates/paimon/src/table/table_scan.rs | 63 ++++++++++++++++++++++----- 1 file changed, 52 insertions(+), 11 deletions(-) diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 18bd9f02..cfe16e1c 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -51,6 +51,7 @@ use crate::table::source::{ }; use crate::table::ScanTrace; use futures::{StreamExt, TryStreamExt}; +use indexmap::IndexMap; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -542,7 +543,20 @@ impl LimitPushdownAccumulator { } } -type BucketDataFileGroups = HashMap<(Vec, i32), (i32, Vec)>; +type BucketDataFileGroups = IndexMap<(Vec, i32), (i32, Vec)>; + +fn group_data_files_by_partition_bucket(entries: Vec) -> BucketDataFileGroups { + let mut groups = BucketDataFileGroups::with_capacity(entries.len()); + for entry in entries { + let (partition, bucket, total_buckets, file) = entry.into_parts(); + groups + .entry((partition, bucket)) + .or_insert_with(|| (total_buckets, Vec::new())) + .1 + .push(file); + } + groups +} #[derive(Clone, Copy)] struct GlobalIndexScanSettings { @@ -1752,14 +1766,7 @@ impl<'a> PaimonTableScan<'a> { } // Group by (partition, bucket), decomposing entries to avoid cloning partition. - let mut groups: BucketDataFileGroups = HashMap::with_capacity(entries.len()); - for e in entries { - let (partition, bucket, total_buckets, file) = e.into_parts(); - let entry = groups - .entry((partition, bucket)) - .or_insert_with(|| (total_buckets, Vec::new())); - entry.1.push(file); - } + let groups = group_data_files_by_partition_bucket(entries); let snapshot_id = snapshot.id(); let base_path = table_path.trim_end_matches('/'); @@ -2005,8 +2012,9 @@ impl<'a> PaimonTableScan<'a> { mod tests { use super::{ data_evolution_row_range_groups, data_file_overlaps_row_range_index, - manifest_file_overlaps_row_range_index, prune_data_evolution_group_by_read_fields, - retain_index_manifest_entry, retain_manifest_entry_row_ranges, retain_manifest_row_ranges, + group_data_files_by_partition_bucket, manifest_file_overlaps_row_range_index, + prune_data_evolution_group_by_read_fields, retain_index_manifest_entry, + retain_manifest_entry_row_ranges, retain_manifest_row_ranges, should_skip_level_zero_for_scan, split_row_ranges_for_files, LimitPushdownAccumulator, PaimonTableScan, RowRangeIndex, TableScan, }; @@ -2301,6 +2309,39 @@ mod tests { files.iter().map(|file| file.file_name.as_str()).collect() } + #[test] + fn test_partition_bucket_groups_preserve_manifest_order() { + let entry = |partition: &[u8], name: &str| { + ManifestEntry::new( + FileKind::Add, + partition.to_vec(), + 0, + 1, + make_evo_file(name, 1, 1, 1, None), + 2, + ) + }; + let groups = group_data_files_by_partition_bucket(vec![ + entry(b"b", "b-1.parquet"), + entry(b"a", "a.parquet"), + entry(b"b", "b-2.parquet"), + ]); + + let ordered = groups + .iter() + .map(|((partition, _), (_, files))| { + (partition.as_slice(), file_names_from_files(files)) + }) + .collect::>(); + assert_eq!( + ordered, + vec![ + (b"b".as_slice(), vec!["b-1.parquet", "b-2.parquet"]), + (b"a".as_slice(), vec!["a.parquet"]), + ] + ); + } + #[test] fn test_merge_manifest_entries_keeps_in_place_upgraded_file() { // Reproduces a single-run compaction "upgrade": the SAME file name is From 665d3c3fb8ea976adc5ce819efdf160d9069bf02 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 7 Aug 2026 08:37:34 -0700 Subject: [PATCH 2/2] fix(scan): mirror Java partition and bucket order --- crates/paimon/src/table/table_scan.rs | 42 +++++++++++++++++---------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index cfe16e1c..38d64381 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -543,18 +543,28 @@ impl LimitPushdownAccumulator { } } -type BucketDataFileGroups = IndexMap<(Vec, i32), (i32, Vec)>; +type PartitionDataFileGroups = IndexMap, IndexMap)>>; +type BucketDataFileGroup = ((Vec, i32), (i32, Vec)); -fn group_data_files_by_partition_bucket(entries: Vec) -> BucketDataFileGroups { - let mut groups = BucketDataFileGroups::with_capacity(entries.len()); +fn group_data_files_by_partition_bucket(entries: Vec) -> Vec { + let mut partitions = PartitionDataFileGroups::new(); for entry in entries { let (partition, bucket, total_buckets, file) = entry.into_parts(); - groups - .entry((partition, bucket)) + partitions + .entry(partition) + .or_default() + .entry(bucket) .or_insert_with(|| (total_buckets, Vec::new())) .1 .push(file); } + + let mut groups = Vec::new(); + for (partition, buckets) in partitions { + for (bucket, files) in buckets { + groups.push(((partition.clone(), bucket), files)); + } + } groups } @@ -2311,33 +2321,35 @@ mod tests { #[test] fn test_partition_bucket_groups_preserve_manifest_order() { - let entry = |partition: &[u8], name: &str| { + let entry = |partition: &[u8], bucket: i32, name: &str| { ManifestEntry::new( FileKind::Add, partition.to_vec(), - 0, - 1, + bucket, + 2, make_evo_file(name, 1, 1, 1, None), 2, ) }; let groups = group_data_files_by_partition_bucket(vec![ - entry(b"b", "b-1.parquet"), - entry(b"a", "a.parquet"), - entry(b"b", "b-2.parquet"), + entry(b"b", 1, "b-1.parquet"), + entry(b"a", 0, "a.parquet"), + entry(b"b", 0, "b-0.parquet"), + entry(b"b", 1, "b-1-next.parquet"), ]); let ordered = groups .iter() - .map(|((partition, _), (_, files))| { - (partition.as_slice(), file_names_from_files(files)) + .map(|((partition, bucket), (_, files))| { + (partition.as_slice(), *bucket, file_names_from_files(files)) }) .collect::>(); assert_eq!( ordered, vec![ - (b"b".as_slice(), vec!["b-1.parquet", "b-2.parquet"]), - (b"a".as_slice(), vec!["a.parquet"]), + (b"b".as_slice(), 1, vec!["b-1.parquet", "b-1-next.parquet"]), + (b"b".as_slice(), 0, vec!["b-0.parquet"]), + (b"a".as_slice(), 0, vec!["a.parquet"]), ] ); }