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..5536026126e8 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 @@ -298,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_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 dbb0a756cc01..e36f6c8bcba6 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 @@ def test_matches_reference_grouping_on_random_inputs(self): _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()