From da095067630bdd871486d870cdeec578458cbeb3 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sat, 15 Aug 2026 19:34:10 -0700 Subject: [PATCH 1/3] [python] Preserve manifest order in Data Evolution splits --- .../scanner/data_evolution_split_generator.py | 22 ++------ .../data_evolution_split_generator_test.py | 55 +++++++++++++++++++ 2 files changed, 60 insertions(+), 17 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 fea6ab868777..60e457e947d5 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 @@ def create_splits(self, file_entries: List[ManifestEntry]) -> List[Split]: """ 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 @@ def weight_func(file_list: List[DataFileMeta]) -> int: 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 @@ def weight_func(file_list: List[DataFileMeta]) -> int: ] 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 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 dbb0a756cc01..cbec18af0bcf 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,56 @@ def test_matches_reference_grouping_on_random_inputs(self): _grouping(_reference_split(files))) +class SplitOrderTest(unittest.TestCase): + @staticmethod + def _entry(name, sequence): + 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=[], + first_row_id=0, + ) + return ManifestEntry( + kind=0, + partition=empty_row, + bucket=0, + total_buckets=1, + file=file, + ) + + def test_preserves_manifest_order_within_row_id_group(self): + class _Options: + options = {} + + class _Table: + table_path = '/table' + options = _Options() + + entries = [ + self._entry('a.parquet', 1), + self._entry('b.parquet', 3), + self._entry('c.parquet', 2), + ] + splits = DataEvolutionSplitGenerator( + _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], + ) + + if __name__ == "__main__": unittest.main() From 50152fe5f80dfa8bcfbe83e35ac9ea666d067f4e Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sat, 15 Aug 2026 20:13:43 -0700 Subject: [PATCH 2/3] [python] Preserve file order after row-range filtering --- .../scanner/data_evolution_split_generator.py | 10 ++++- .../data_evolution_split_generator_test.py | 42 +++++++++++++++---- 2 files changed, 42 insertions(+), 10 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 60e457e947d5..c89987e3e23a 100644 --- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py +++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py @@ -286,8 +286,14 @@ def _filter_files_by_row_ranges(partitioned_files: defaultdict, row_ranges: List filtered_blob_entries.append(entry) break - # Combine filtered non-blob and blob files - filtered_entries = filtered_non_blob_entries + filtered_blob_entries + kept_files = { + entry.file.file_name + for entry in filtered_non_blob_entries + filtered_blob_entries + } + filtered_entries = [ + entry for entry in file_entries + if entry.file.file_name in kept_files + ] 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 cbec18af0bcf..26d6dfd6ee1a 100644 --- a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py +++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py @@ -117,6 +117,15 @@ def test_matches_reference_grouping_on_random_inputs(self): class SplitOrderTest(unittest.TestCase): + class _Options: + options = {} + + class _Table: + table_path = '/table' + options = None + + _Table.options = _Options() + @staticmethod def _entry(name, sequence): empty_row = GenericRow([], []) @@ -145,20 +154,13 @@ def _entry(name, sequence): ) def test_preserves_manifest_order_within_row_id_group(self): - class _Options: - options = {} - - class _Table: - table_path = '/table' - options = _Options() - entries = [ self._entry('a.parquet', 1), self._entry('b.parquet', 3), self._entry('c.parquet', 2), ] splits = DataEvolutionSplitGenerator( - _Table(), target_split_size=1024, open_file_cost=0 + self._Table(), target_split_size=1024, open_file_cost=0 ).create_splits(entries) self.assertEqual( @@ -166,6 +168,30 @@ class _Table: [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], + ) + if __name__ == "__main__": unittest.main() From 7aa3172155ec310b1c83c0a9ec73f3e149c3149c Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sun, 16 Aug 2026 00:43:34 -0700 Subject: [PATCH 3/3] [python] Preserve exact entries in slice filtering --- .../scanner/data_evolution_split_generator.py | 6 ++-- .../data_evolution_split_generator_test.py | 35 +++++++++++++++++-- 2 files changed, 36 insertions(+), 5 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 c89987e3e23a..5536026126e8 100644 --- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py +++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py @@ -286,13 +286,13 @@ def _filter_files_by_row_ranges(partitioned_files: defaultdict, row_ranges: List filtered_blob_entries.append(entry) break - kept_files = { - entry.file.file_name + kept_entries = { + id(entry) for entry in filtered_non_blob_entries + filtered_blob_entries } filtered_entries = [ entry for entry in file_entries - if entry.file.file_name in kept_files + if id(entry) in kept_entries ] if 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 26d6dfd6ee1a..e36f6c8bcba6 100644 --- a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py +++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py @@ -127,7 +127,7 @@ class _Table: _Table.options = _Options() @staticmethod - def _entry(name, sequence): + def _entry(name, sequence, first_row_id=0, external_path=None): empty_row = GenericRow([], []) empty_stats = SimpleStats(empty_row, empty_row, []) file = DataFileMeta.create( @@ -143,7 +143,8 @@ def _entry(name, sequence): schema_id=0, level=0, extra_files=[], - first_row_id=0, + external_path=external_path, + first_row_id=first_row_id, ) return ManifestEntry( kind=0, @@ -192,6 +193,36 @@ def test_slice_and_shard_preserve_blob_manifest_order(self): [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()