diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 18bd9f02..38d64381 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,30 @@ impl LimitPushdownAccumulator { } } -type BucketDataFileGroups = HashMap<(Vec, i32), (i32, Vec)>; +type PartitionDataFileGroups = IndexMap, IndexMap)>>; +type BucketDataFileGroup = ((Vec, i32), (i32, Vec)); + +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(); + 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 +} #[derive(Clone, Copy)] struct GlobalIndexScanSettings { @@ -1752,14 +1776,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 +2022,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 +2319,41 @@ 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], bucket: i32, name: &str| { + ManifestEntry::new( + FileKind::Add, + partition.to_vec(), + bucket, + 2, + make_evo_file(name, 1, 1, 1, None), + 2, + ) + }; + let groups = group_data_files_by_partition_bucket(vec![ + 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, bucket), (_, files))| { + (partition.as_slice(), *bucket, file_names_from_files(files)) + }) + .collect::>(); + assert_eq!( + ordered, + vec![ + (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"]), + ] + ); + } + #[test] fn test_merge_manifest_entries_keeps_in_place_upgraded_file() { // Reproduces a single-run compaction "upgrade": the SAME file name is