[python] Preserve manifest order in Data Evolution splits

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

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()