[python] Cover native dynamic PK events and compacted first-row scans

paimon-python/pypaimon/read/table_scan.py

@@ -180,8 +180,9 @@ def _native_plan_supported_impl(self) -> bool:
             return False
         if self.table.options.query_auth_enabled:
             return False
-        # Java batch first-row reads skip L0. Scans including L0 still need
-        # first-row overlap packing that Rust does not currently provide.
+        # Ordinary Java first-row batch scans skip L0. The clustering override
+        # can combine first-row with DV merge-on-read, but its files need a
+        # separate audit because they may be sorted by non-primary-key columns.
         if (self.table.options.merge_engine() == 'first-row'
                 and not fs.skip_level0 and not fs.is_streaming):
             return False

paimon-python/pypaimon/tests/native_plan_capabilities_test.py

@@ -314,6 +314,42 @@ def test_primary_key_deletion_vectors_preserve_compacted_rows(self):
         self.assertEqual(plan.splits()[0].data_deletion_files[0].cardinality, 2)
         self._assert_parity(table.copy({'scan.snapshot-id': '1'}), rows, 1)
 
+    def test_first_row_compacted_runs_with_overlapping_key_ranges(self):
+        for bucket in ('1', '-1'):
+            with self.subTest(bucket=bucket):
+                table = self._create('first_row_' + bucket, {
+                    'bucket': bucket, 'merge-engine': 'first-row',
+                    'source.split.target-size': '1b',
+                    'source.split.open-file-cost': '1b',
+                }, primary_keys=['k'])
+                expected = [{'k': k, 'v': 'v%d' % k} for k in range(1, 5)]
+                for level, keys in ((1, (1, 3)), (2, (2, 4))):
+                    builder = table.new_batch_write_builder()
+                    writer, commit = builder.new_write(), builder.new_commit()
+                    try:
+                        writer.write_arrow(pa.Table.from_pylist(
+                            [expected[key - 1] for key in keys], schema=self.schema))
+                        messages = writer.prepare_commit()
+                        # Each run already contains unique first rows. Their
+                        # ranges overlap, but their actual keys are disjoint.
+                        for message in messages:
+                            message.new_files = [replace(file, level=level)
+                                                 for file in message.new_files]
+                        commit.commit(messages)
+                    finally:
+                        writer.close()
+                        commit.close()
+                self._write(table, [{'k': 1, 'v': 'later'}, {'k': 5, 'v': 'pending'}])
+                plan = self._assert_parity(table, expected, 3)
+                self.assertTrue(all(split.raw_convertible for split in plan.splits()))
+                self.assertEqual({file.level for split in plan.splits()
+                                  for file in split.files}, {1, 2})
+                pb = table.new_read_builder().new_predicate_builder()
+                self._assert_parity(table, [expected[1]], 3, predicate=pb.equal('v', 'v2'))
+                self._assert_parity(table, [], 3, predicate=pb.equal('v', 'later'))
+                self._assert_parity(table.copy({'scan.snapshot-id': '1'}),
+                                    [expected[0], expected[2]], 1)
+
     @unittest.skipUnless(native_version_at_least(0, 4, 0),
                          'pypaimon-rust>=0.4.0 required for native DV scans')
     def test_external_deletion_vector_path_is_preserved(self):

paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py

@@ -17,6 +17,7 @@
 
 """Real bucket growth and cross-partition updates through native planning."""
 
+import json
 from unittest.mock import patch
 
 import pyarrow as pa
@@ -75,6 +76,11 @@ def read(table, native, predicate=None, shard=None, limit=None):
         writer.write_arrow(pa.RecordBatch.from_pylist(rows, schema=schema))
         builder.new_commit().commit(writer.prepare_commit())
         table = catalog.get_table('default.t')
+        # Give incremental windows deterministic boundaries.
+        path = table.snapshot_manager().get_snapshot_path(snapshot_id)
+        snapshot = json.loads(table.file_io.read_file_utf8(path))
+        snapshot['timeMillis'] = snapshot_id * 100
+        table.file_io.write_file(path, json.dumps(snapshot), overwrite=True)
         pb = table.new_read_builder().new_predicate_builder()
         for predicate in (None, pb.equal('id', 1), pb.equal('p', 'a'), pb.equal('v', 'updated')):
             for shard in (None, (0, 2), (1, 2)):
@@ -99,3 +105,30 @@ def read(table, native, predicate=None, shard=None, limit=None):
         old_plan, old_rows = read(table.copy({'scan.snapshot-id': '1'}), native)
         assert old_plan.snapshot_id == 1
         assert old_rows == initial
+
+    # Java DeleteExistingProcessor emits DELETE in the old partition with
+    # the incoming non-partition values. Batch merging hides this distinction;
+    # incremental readers must preserve both the partition and the row kind.
+    events = [(row['id'], row['p'], row['v'], 0) for row in initial + updates]
+    if cross_partition:
+        events.append((1, 'a', 'moved', 3))
+    for native in (False, True):
+        for partition in (None, 'a', 'b'):
+            builder = table.copy({
+                'scan.native-plan.enabled': str(native).lower(),
+                'incremental-between-timestamp': '0,200',
+            }).new_read_builder()
+            if partition is not None:
+                builder.with_filter(pb.equal('p', partition))
+            scan = builder.new_scan()
+            if native:
+                with patch.object(scan.file_scanner, 'scan', side_effect=AssertionError('native fallback')):
+                    plan = scan.plan()
+            else:
+                plan = scan.plan()
+            assert all(split.is_streaming for split in plan.splits())
+            actual = [(row.get_field(0), row.get_field(1), row.get_field(2),
+                       row.get_row_kind().value)
+                      for row in builder.new_read().to_iterator(plan.splits())]
+            assert sorted(actual) == sorted(
+                event for event in events if partition is None or event[1] == partition)