This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new b27c3005 fix(scan): preserve partition group order (#692)
b27c3005 is described below
commit b27c30054e17ee11f7400bf07a8fd41cf264f08b
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Aug 8 13:04:44 2026 +0800
fix(scan): preserve partition group order (#692)
---
crates/paimon/src/table/table_scan.rs | 75 ++++++++++++++++++++++++++++++-----
1 file changed, 64 insertions(+), 11 deletions(-)
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<u8>, i32), (i32, Vec<DataFileMeta>)>;
+type PartitionDataFileGroups = IndexMap<Vec<u8>, IndexMap<i32, (i32,
Vec<DataFileMeta>)>>;
+type BucketDataFileGroup = ((Vec<u8>, i32), (i32, Vec<DataFileMeta>));
+
+fn group_data_files_by_partition_bucket(entries: Vec<ManifestEntry>) ->
Vec<BucketDataFileGroup> {
+ 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::<Vec<_>>();
+ 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
