This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2bbdb668f0 [python] Cover native dynamic PK events and compacted
first-row scans (#9854)
2bbdb668f0 is described below
commit 2bbdb668f02d37612702b54c20425be75808f1b4
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)
