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 5574d707 fix(scan): preserve Data Evolution file order in row-id
groups (#717)
5574d707 is described below
commit 5574d707f023a10099085c30165508fc321e3190
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 16 22:46:20 2026 +0800
fix(scan): preserve Data Evolution file order in row-id groups (#717)
---
crates/paimon/src/table/stats_filter.rs | 46 ++++++++++++++++++++-------------
crates/paimon/src/table/table_scan.rs | 10 +++----
2 files changed, 33 insertions(+), 23 deletions(-)
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index 8e2f06b0..010a2344 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -571,29 +571,39 @@ fn data_evolution_predicate_must_match(
/// Groups data files by overlapping `row_id_range` for data evolution.
///
-/// Files are sorted by `(first_row_id, -max_sequence_number)`. Files whose
row ID ranges
-/// overlap are merged into the same group (they contain different columns for
the same rows).
-/// Files without `first_row_id` become their own group.
+/// Groups are sorted by row ID range. Files whose row ID ranges overlap are
merged into
+/// the same group while preserving their input order. Files without
`first_row_id` become
+/// their own group.
///
-/// Reference:
[DataEvolutionSplitGenerator](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/DataEvolutionSplitGenerator.java)
-pub(crate) fn group_by_overlapping_row_id(mut files: Vec<DataFileMeta>) ->
Vec<Vec<DataFileMeta>> {
- files.sort_by(|a, b| {
- let a_row_id = a.first_row_id.unwrap_or(i64::MIN);
- let b_row_id = b.first_row_id.unwrap_or(i64::MIN);
- a_row_id
- .cmp(&b_row_id)
- .then_with(|| b.max_sequence_number.cmp(&a.max_sequence_number))
+/// Reference:
[DataEvolutionSplitGenerator](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java)
+pub(crate) fn group_by_overlapping_row_id(files: Vec<DataFileMeta>) ->
Vec<Vec<DataFileMeta>> {
+ let mut indexed_files: Vec<(usize, DataFileMeta)> =
files.into_iter().enumerate().collect();
+ indexed_files.sort_by(|(a_index, a), (b_index, b)| {
+ match (a.row_id_range(), b.row_id_range()) {
+ (None, None) => a_index.cmp(b_index),
+ (None, Some(_)) => std::cmp::Ordering::Less,
+ (Some(_), None) => std::cmp::Ordering::Greater,
+ (Some((a_start, a_end)), Some((b_start, b_end))) => a_start
+ .cmp(&b_start)
+ .then_with(|| a_end.cmp(&b_end))
+ .then_with(|| a_index.cmp(b_index)),
+ }
});
let mut result: Vec<Vec<DataFileMeta>> = Vec::new();
- let mut current_group: Vec<DataFileMeta> = Vec::new();
+ let mut current_group: Vec<(usize, DataFileMeta)> = Vec::new();
let mut current_range_end: i64 = i64::MIN;
- for file in files {
+ let restore_input_order = |mut group: Vec<(usize, DataFileMeta)>| {
+ group.sort_by_key(|(index, _)| *index);
+ group.into_iter().map(|(_, file)| file).collect()
+ };
+
+ for (index, file) in indexed_files {
match file.row_id_range() {
None => {
if !current_group.is_empty() {
- result.push(std::mem::take(&mut current_group));
+ result.push(restore_input_order(std::mem::take(&mut
current_group)));
current_range_end = i64::MIN;
}
result.push(vec![file]);
@@ -603,17 +613,17 @@ pub(crate) fn group_by_overlapping_row_id(mut files:
Vec<DataFileMeta>) -> Vec<V
if end > current_range_end {
current_range_end = end;
}
- current_group.push(file);
+ current_group.push((index, file));
} else {
- result.push(std::mem::take(&mut current_group));
+ result.push(restore_input_order(std::mem::take(&mut
current_group)));
current_range_end = end;
- current_group.push(file);
+ current_group.push((index, file));
}
}
}
}
if !current_group.is_empty() {
- result.push(current_group);
+ result.push(restore_input_order(current_group));
}
result
}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index e4cd60ef..35b8b1a8 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -2899,7 +2899,7 @@ mod tests {
make_evo_file("b", 10, 100, 2, None),
];
let groups = group_by_overlapping_row_id(files);
- assert_eq!(file_names(&groups), vec![vec!["b"], vec!["a"]]);
+ assert_eq!(file_names(&groups), vec![vec!["a"], vec!["b"]]);
}
#[test]
@@ -3247,8 +3247,8 @@ mod tests {
#[test]
fn test_group_by_overlapping_row_id_non_overlapping() {
let files = vec![
- make_evo_file("a", 10, 100, 1, Some(0)),
make_evo_file("b", 10, 100, 2, Some(100)),
+ make_evo_file("a", 10, 100, 1, Some(0)),
];
let groups = group_by_overlapping_row_id(files);
assert_eq!(groups.len(), 2);
@@ -3266,12 +3266,12 @@ mod tests {
let groups = group_by_overlapping_row_id(files);
assert_eq!(
file_names(&groups),
- vec![vec!["c"], vec!["b", "a"], vec!["d"]]
+ vec![vec!["c"], vec!["a", "b"], vec!["d"]]
);
}
#[test]
- fn test_group_by_overlapping_row_id_sorted_by_seq() {
+ fn test_group_by_overlapping_row_id_preserves_input_order() {
let files = vec![
make_evo_file("a", 10, 100, 1, Some(0)),
make_evo_file("b", 10, 100, 3, Some(0)),
@@ -3279,7 +3279,7 @@ mod tests {
];
let groups = group_by_overlapping_row_id(files);
assert_eq!(groups.len(), 1);
- assert_eq!(file_names(&groups), vec![vec!["b", "c", "a"]]);
+ assert_eq!(file_names(&groups), vec![vec!["a", "b", "c"]]);
}
#[test]
