This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f3862f6b4e5774a244b99104122d6767c144792b Author: Jingsong Lee <[email protected]> AuthorDate: Wed Sep 16 09:08:54 2026 +0800 [python] Cover native dynamic PK events and compacted first-row scans (#9854) --- paimon-python/pypaimon/read/table_scan.py | 5 +-- .../tests/native_plan_capabilities_test.py | 36 ++++++++++++++++++++++ .../tests/native_plan_dynamic_bucket_test.py | 33 ++++++++++++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/read/table_scan.py b/paimon-python/pypaimon/read/table_scan.py index 9a07b5ab9d..7e4889da51 100755 --- a/paimon-python/pypaimon/read/table_scan.py +++ b/paimon-python/pypaimon/read/table_scan.py @@ -180,8 +180,9 @@ class TableScan: 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 diff --git a/paimon-python/pypaimon/tests/native_plan_capabilities_test.py b/paimon-python/pypaimon/tests/native_plan_capabilities_test.py index 2a35b75b7c..867b3261cd 100644 --- a/paimon-python/pypaimon/tests/native_plan_capabilities_test.py +++ b/paimon-python/pypaimon/tests/native_plan_capabilities_test.py @@ -314,6 +314,42 @@ class NativePlanCapabilitiesTest(unittest.TestCase): 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): diff --git a/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py b/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py index 1fe383e773..27ca00ef6f 100644 --- a/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py +++ b/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 test_native_pk_bucket_growth_and_partition_migration(tmp_path, cross_partiti 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 test_native_pk_bucket_growth_and_partition_migration(tmp_path, cross_partiti 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)
