This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 12635dffc8 [python] Preserve manifest order in Data Evolution splits
(#9242)
12635dffc8 is described below
commit 12635dffc8d75a8c2698345ea943234650bbe965
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 16 16:10:29 2026 +0800
[python] Preserve manifest order in Data Evolution splits (#9242)
---
.../read/scanner/data_evolution_split_generator.py | 32 +++---
.../tests/data_evolution_split_generator_test.py | 112 +++++++++++++++++++++
2 files changed, 125 insertions(+), 19 deletions(-)
diff --git
a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
index fea6ab8687..5536026126 100644
--- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
@@ -51,20 +51,8 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
"""
Create splits for data evolution tables.
"""
- def sort_key(manifest_entry: ManifestEntry) -> tuple:
- first_row_id = (
- manifest_entry.file.first_row_id
- if manifest_entry.file.first_row_id is not None
- else float('-inf')
- )
- is_blob = 1 if
DataFileMeta.is_blob_file(manifest_entry.file.file_name) else 0
- max_seq = manifest_entry.file.max_sequence_number
- return first_row_id, is_blob, -max_seq
-
- sorted_entries = sorted(file_entries, key=sort_key)
-
partitioned_files = defaultdict(list)
- for entry in sorted_entries:
+ for entry in file_entries:
partitioned_files[(tuple(entry.partition.values),
entry.bucket)].append(entry)
slice_row_ranges = None # Row ID ranges for slice-based filtering
@@ -79,11 +67,11 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
return max(sum(f.file_size for f in file_list),
self.open_file_cost)
splits = []
- for key, sorted_entries_list in partitioned_files.items():
- if not sorted_entries_list:
+ for key, entries_list in partitioned_files.items():
+ if not entries_list:
continue
- data_files: List[DataFileMeta] = [e.file for e in
sorted_entries_list]
+ data_files: List[DataFileMeta] = [e.file for e in entries_list]
# Split files by firstRowId for data evolution
split_by_row_id = self._split_by_row_id(data_files)
@@ -109,7 +97,7 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
]
splits += self._build_split_from_pack_for_data_evolution(
- flatten_packed_files, packed_files, sorted_entries_list
+ flatten_packed_files, packed_files, entries_list
)
# merge slice_row_ranges and self.row_ranges
@@ -298,8 +286,14 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
filtered_blob_entries.append(entry)
break
- # Combine filtered non-blob and blob files
- filtered_entries = filtered_non_blob_entries +
filtered_blob_entries
+ kept_entries = {
+ id(entry)
+ for entry in filtered_non_blob_entries + filtered_blob_entries
+ }
+ filtered_entries = [
+ entry for entry in file_entries
+ if id(entry) in kept_entries
+ ]
if filtered_entries:
filtered_partitioned_files[key] = filtered_entries
diff --git
a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
index dbb0a756cc..e36f6c8bcb 100644
--- a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
@@ -18,7 +18,11 @@
import random
import unittest
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.read.scanner.data_evolution_split_generator import
DataEvolutionSplitGenerator
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.utils.range import Range
@@ -112,5 +116,113 @@ class SplitByRowIdEquivalenceTest(unittest.TestCase):
_grouping(_reference_split(files)))
+class SplitOrderTest(unittest.TestCase):
+ class _Options:
+ options = {}
+
+ class _Table:
+ table_path = '/table'
+ options = None
+
+ _Table.options = _Options()
+
+ @staticmethod
+ def _entry(name, sequence, first_row_id=0, external_path=None):
+ empty_row = GenericRow([], [])
+ empty_stats = SimpleStats(empty_row, empty_row, [])
+ file = DataFileMeta.create(
+ file_name=name,
+ file_size=1,
+ row_count=10,
+ min_key=empty_row,
+ max_key=empty_row,
+ key_stats=empty_stats,
+ value_stats=empty_stats,
+ min_sequence_number=sequence,
+ max_sequence_number=sequence,
+ schema_id=0,
+ level=0,
+ extra_files=[],
+ external_path=external_path,
+ first_row_id=first_row_id,
+ )
+ return ManifestEntry(
+ kind=0,
+ partition=empty_row,
+ bucket=0,
+ total_buckets=1,
+ file=file,
+ )
+
+ def test_preserves_manifest_order_within_row_id_group(self):
+ entries = [
+ self._entry('a.parquet', 1),
+ self._entry('b.parquet', 3),
+ self._entry('c.parquet', 2),
+ ]
+ splits = DataEvolutionSplitGenerator(
+ self._Table(), target_split_size=1024, open_file_cost=0
+ ).create_splits(entries)
+
+ self.assertEqual(
+ ['a.parquet', 'b.parquet', 'c.parquet'],
+ [file.file_name for file in splits[0].files],
+ )
+
+ def test_slice_and_shard_preserve_blob_manifest_order(self):
+ entries = [
+ self._entry('a.blob', 1),
+ self._entry('b.parquet', 2),
+ self._entry('c.blob', 3),
+ ]
+ expected = ['a.blob', 'b.parquet', 'c.blob']
+
+ generators = [
+ DataEvolutionSplitGenerator(
+ self._Table(), target_split_size=1024, open_file_cost=0
+ ).with_slice(0, 5),
+ DataEvolutionSplitGenerator(
+ self._Table(), target_split_size=1024, open_file_cost=0
+ ).with_shard(0, 2),
+ ]
+ for generator in generators:
+ with self.subTest(generator=type(generator).__name__):
+ splits = generator.create_splits(entries)
+ self.assertEqual(
+ expected,
+ [file.file_name for file in splits[0].files],
+ )
+
+ def test_slice_and_shard_distinguish_same_external_file_name(self):
+ entries = [
+ self._entry(
+ 'same.parquet', 1, first_row_id=0,
+ external_path='s3://bucket-a/data/same.parquet',
+ ),
+ self._entry(
+ 'same.parquet', 2, first_row_id=10,
+ external_path='s3://bucket-b/data/same.parquet',
+ ),
+ ]
+ expected = ['s3://bucket-a/data/same.parquet']
+
+ generators = [
+ DataEvolutionSplitGenerator(
+ self._Table(), target_split_size=1024, open_file_cost=0
+ ).with_slice(0, 10),
+ DataEvolutionSplitGenerator(
+ self._Table(), target_split_size=1024, open_file_cost=0
+ ).with_shard(0, 2),
+ ]
+ for generator in generators:
+ with self.subTest(generator=type(generator).__name__):
+ splits = generator.create_splits(entries)
+ self.assertEqual(
+ expected,
+ [file.external_path for split in splits
+ for file in split.files],
+ )
+
+
if __name__ == "__main__":
unittest.main()
