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 0c1503fec7 [python] Align batch scan semantics with Java and extend
native planning (#9839)
0c1503fec7 is described below
commit 0c1503fec7ee919f0f895f5a9b6d1be1466b9cba
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 17:04:16 2026 +0800
[python] Align batch scan semantics with Java and extend native planning
(#9839)
---
paimon-python/README.md | 17 ++-
.../pypaimon/common/options/core_options.py | 13 ++
.../pypaimon/manifest/manifest_file_manager.py | 5 +-
paimon-python/pypaimon/read/native_plan.py | 2 +
.../pypaimon/read/scanner/file_scanner.py | 12 +-
paimon-python/pypaimon/read/table_scan.py | 37 ++---
.../pypaimon/tests/batch_scan_level_filter_test.py | 154 +++++++++++++++++++++
.../tests/manifest/manifest_manager_test.py | 18 +++
.../tests/native_plan_dynamic_bucket_test.py | 101 ++++++++++++++
.../pypaimon/tests/native_plan_integration_test.py | 5 +-
paimon-python/pypaimon/tests/native_plan_test.py | 11 +-
.../tests/schema_evolution_pk_read_test.py | 9 +-
.../pypaimon/tests/table_query_auth_test.py | 1 +
paimon-python/pypaimon/tests/test_first_row_e2e.py | 9 +-
.../pypaimon/tests/test_partial_update_e2e.py | 4 +-
15 files changed, 354 insertions(+), 44 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 2667bb5c45..76155304b2 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -78,11 +78,18 @@ the selected APPEND deltas into one plan, including merging
primary-key versions
across commits. Other commit kinds are excluded, and the ending snapshot
supplies
snapshot metadata and deletion vectors even if it contributes no APPEND files.
-Chunk shuffle, query authorization, first-row merge, deletion-vector
merge-on-read,
-dynamic or cross-partition primary-key buckets, and scored or primary-key
-global-index results still use the Python planner. Rust also rejects
floating-point
-partition-directory formatting; these scans fall back to Python. Native
planning
-remains optional and is disabled by default.
+Dynamic and cross-partition primary-key buckets support native planning,
including
+bucket sharding. Cross-partition key migration is maintained by the writer's
index.
+Batch first-row scans follow Java and exclude un-compacted level-0 files; they
can
+use native planning. With deletion vectors, batch scans exclude level 0 unless
+`deletion-vectors.merge-on-read=true`, in which case overlapping key ranges
stay
+together for reader-side merging. Write scans and incremental scans retain
level 0.
+
+Chunk shuffle, query authorization, first-row scans that include level 0, and
scored
+or primary-key global-index results still use the Python planner. Native
planning
+remains optional and is disabled by default. The combined incremental planner's
+window-end deletion-vector behavior described above differs from Java delta
scans,
+which do not attach deletion vectors; full Java incremental parity remains
pending.
# Load LeRobot Dataset v3
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index ce77d21f5a..43a07a36a7 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -638,6 +638,13 @@ class CoreOptions:
.with_description("Whether to enable deletion vectors.")
)
+ DELETION_VECTORS_MERGE_ON_READ: ConfigOption[bool] = (
+ ConfigOptions.key("deletion-vectors.merge-on-read")
+ .boolean_type()
+ .default_value(False)
+ .with_description("Whether batch reads merge level-0 files when
deletion vectors are enabled.")
+ )
+
INDEX_FILE_IN_DATA_FILE_DIR: ConfigOption[bool] = (
ConfigOptions.key("index-file-in-data-file-dir")
.boolean_type()
@@ -1500,6 +1507,12 @@ class CoreOptions:
def deletion_vectors_enabled(self, default=None):
return self.options.get(CoreOptions.DELETION_VECTORS_ENABLED, default)
+ def batch_scan_skip_level0(self):
+ """Match Java CoreOptions.batchScanSkipLevel0."""
+ if self.deletion_vectors_enabled():
+ return not
self.options.get(CoreOptions.DELETION_VECTORS_MERGE_ON_READ)
+ return self.merge_engine() == MergeEngine.FIRST_ROW
+
def index_file_in_data_file_dir(self, default=None):
return self.options.get(CoreOptions.INDEX_FILE_IN_DATA_FILE_DIR,
default)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index af2dec5388..7b438da88f 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -57,14 +57,15 @@ def _read_manifest_records(buffer, early_entry_filter,
partition_filter,
return
blocks = fastavro.block_reader(buffer)
- fields = blocks.writer_schema.get('fields', [])
+ schema = blocks.writer_schema
+ fields = schema.get('fields', []) if isinstance(schema, dict) else []
if (tuple(field.get('name') for field in fields) != _MANIFEST_FIELD_NAMES
or tuple(field.get('type') for field in fields[:5])
!= _MANIFEST_PREFIX_TYPES
or not isinstance(fields[5].get('type'), dict)
or fields[5]['type'].get('type') != 'record'):
# Avro values follow writer field order. Keep the generic reader for
- # historical manifests whose top-level fields were reordered.
+ # reordered fields or a top-level union, as written by Rust.
buffer.seek(0)
for record in fastavro.reader(buffer):
yield record, None, False
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index b3f6ae9079..bac2dd3784 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -142,6 +142,8 @@ def _read_options(table) -> dict:
table.options.source_split_target_size()),
CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): str(
table.options.source_split_open_file_cost()),
+ CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key():
_option_value_to_string(
+
table.options.options.get(CoreOptions.DELETION_VECTORS_MERGE_ON_READ)),
}
table_options = table.options.options
for option in (
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index f8facf435b..a02e745bd6 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -222,7 +222,8 @@ class FileScanner:
manifest_scanner: Callable[[], Tuple[List[ManifestFileMeta],
Optional[Snapshot]]],
predicate: Optional[Predicate] = None,
limit: Optional[int] = None,
- partition_predicate: Optional[Predicate] = None
+ partition_predicate: Optional[Predicate] = None,
+ skip_level0: bool = False,
):
from pypaimon.table.file_store_table import FileStoreTable
@@ -274,6 +275,7 @@ class FileScanner:
self.only_read_real_buckets = options.bucket() ==
BucketMode.POSTPONE_BUCKET.value
self.data_evolution = options.data_evolution_enabled()
self.deletion_vectors_enabled = options.deletion_vectors_enabled()
+ self.skip_level0 = skip_level0
self._global_index_result = None
self._row_ranges = None
self._scanned_snapshot = None
@@ -869,7 +871,7 @@ class FileScanner:
# Apply evolution to stats
if self.table.is_primary_key_table:
- if self.deletion_vectors_enabled and entry.file.level == 0: # do
not read level 0 file
+ if self.skip_level0 and entry.file.level == 0:
return False
if self.primary_key_predicate:
if not self.primary_key_predicate.test_by_simple_stats(
@@ -877,9 +879,9 @@ class FileScanner:
entry.file.row_count
):
return False
- # In DV mode, files within a bucket don't overlap (level 0
excluded above),
- # so we can safely filter by value stats per file.
- if self.deletion_vectors_enabled and self.predicate_for_stats:
+ # Java enables value filtering only when batch scans exclude L0.
+ # With L0 present, pruning an update can expose an older value.
+ if self.skip_level0 and self.predicate_for_stats:
if entry.file.value_stats_cols is None and
entry.file.write_cols is not None:
stats_fields = entry.file.write_cols
else:
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 04edea9a70..d2ee59a4b6 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -46,6 +46,7 @@ _NATIVE_FORWARDED_OPTIONS = frozenset({
CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(),
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(),
CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(),
+ CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(),
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_TIMESTAMP.key(),
@@ -113,11 +114,9 @@ class TableScan:
def _native_plan_supported_impl(self) -> bool:
"""Fall back to the Python scanner for scans native can't carry:
chunk-shuffle, scored or primary-key
- global-index results, first-row merge-engine (Rust drops L0), deletion
- vector merge-on-read, postpone bucket,
+ global-index results, first-row scans which include L0, postpone
bucket,
a primary-key table whose trimmed PK is empty (PK equals the partition
- key; Rust rejects this schema), dynamic
- bucket / cross-partition PK tables (unconfirmed Rust parity), a stale
+ key; Rust rejects this schema), a stale
schema without time travel, removed copy() options which Rust cannot
represent, unsupported time travel selectors,
query auth, a missing/old
@@ -143,13 +142,6 @@ class TableScan:
# 0.4.0 includes Python-written DV decoding and legacy bucket
paths.
if not native_version_at_least(0, 4, 0):
return False
- # Python DV scans skip L0; Rust can include L0 when this option is
- # enabled. Keep that mode on the Python planner until aligned.
- table_options = self.table.options.options.to_map()
- merge_on_read = table_options.get(
- 'deletion-vectors.merge-on-read', False)
- if str(merge_on_read).lower() == 'true':
- return False
if getattr(fs, 'data_evolution', False):
if (getattr(fs, 'idx_of_this_subtask', None) is not None
and not native_method_available('TableScan',
'with_row_position_shard')):
@@ -184,17 +176,17 @@ class TableScan:
database_name = self.table.identifier.get_database_name()
if not database_name or database_name == UNKNOWN_DATABASE or '.' in
database_name:
return False
- if self.table.options.query_auth_enabled \
- or self.table.options.merge_engine() == 'first-row':
+ 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.
+ if (self.table.options.merge_engine() == 'first-row'
+ and not fs.skip_level0):
return False
# Rust rejects schemas whose primary keys are all partition keys.
if getattr(self.table, 'is_primary_key_table', False) \
and not self.table.trimmed_primary_keys:
return False
- # Dynamic-bucket / cross-partition PK: Rust parity unconfirmed -> fall
back.
- from pypaimon.table.bucket_mode import BucketMode
- if self.table.bucket_mode() in (BucketMode.HASH_DYNAMIC,
BucketMode.CROSS_PARTITION):
- return False
options = self.table.options.options
if (options.contains_key(CoreOptions.SCAN_WATERMARK.key())
and not native_version_at_least(0, 4)):
@@ -325,7 +317,13 @@ class TableScan:
def plan_for_write(self) -> Plan:
if self.__auth_query() is not None:
raise TableNoPermissionException(self.table.identifier)
- return self.file_scanner.scan()
+ # Writer restore/overwrite must see L0 even when batch reads hide it.
+ skip_level0 = self.file_scanner.skip_level0
+ try:
+ self.file_scanner.skip_level0 = False
+ return self.file_scanner.scan()
+ finally:
+ self.file_scanner.skip_level0 = skip_level0
def __auth_query(self):
return resolve_auth_result(self._query_auth_fn, self._read_type)
@@ -430,6 +428,7 @@ class TableScan:
self.predicate,
self.limit,
partition_predicate=self.partition_predicate,
+ skip_level0=False,
)
if has_time_travel:
@@ -449,6 +448,7 @@ class TableScan:
self.predicate,
self.limit,
partition_predicate=self.partition_predicate,
+ skip_level0=self.table.options.batch_scan_skip_level0(),
)
def all_manifests():
@@ -461,6 +461,7 @@ class TableScan:
self.predicate,
self.limit,
partition_predicate=self.partition_predicate,
+ skip_level0=self.table.options.batch_scan_skip_level0(),
)
def with_shard(self, idx_of_this_subtask, number_of_para_subtasks) ->
'TableScan':
diff --git a/paimon-python/pypaimon/tests/batch_scan_level_filter_test.py
b/paimon-python/pypaimon/tests/batch_scan_level_filter_test.py
new file mode 100644
index 0000000000..d4daf96d74
--- /dev/null
+++ b/paimon-python/pypaimon/tests/batch_scan_level_filter_test.py
@@ -0,0 +1,154 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Java AbstractBatchTableScan L0 filtering, including native planning."""
+
+import json
+from dataclasses import replace
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+
+
+SCHEMA = pa.schema([('id', pa.int64()), ('v', pa.string())])
+
+
[email protected](params=[False, pytest.param(
+ True, marks=[pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='native planning required')])],
+ ids=['python', 'native'])
+def native(request):
+ return request.param
+
+
+def _table(tmp_path, engine, dv, mor):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ SCHEMA, primary_keys=['id'], options={
+ 'bucket': '1', 'merge-engine': engine,
+ 'deletion-vectors.enabled': str(dv).lower(),
+ 'deletion-vectors.merge-on-read': str(mor).lower(),
+ 'source.split.target-size': '1 b',
+ 'source.split.open-file-cost': '1 b',
+ }), False)
+ return catalog.get_table('default.t')
+
+
+def _write(table, rows, timestamp, level=0, overwrite=False):
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=SCHEMA))
+ messages = writer.prepare_commit()
+ # A unique-key file marked L1 models the output of Java compaction.
+ # Subsequent unmerged updates stay at L0.
+ 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()
+ manager = table.snapshot_manager()
+ path = manager.get_snapshot_path(manager.get_latest_snapshot().id)
+ snapshot = json.loads(table.file_io.read_file_utf8(path))
+ snapshot['timeMillis'] = timestamp
+ table.file_io.write_file(path, json.dumps(snapshot), overwrite=True)
+
+
+def _read(table, native, predicate=None):
+ builder = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ if predicate is not None:
+ builder.with_filter(predicate)
+ 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()
+ return plan,
sorted(builder.new_read().to_arrow(plan.splits()).to_pylist(), key=lambda row:
row['id'])
+
+
[email protected]('engine,dv,mor,skip_l0', [
+ ('first-row', False, False, True),
+ ('deduplicate', True, False, True),
+ ('deduplicate', True, True, False),
+ ('deduplicate', False, False, False),
+])
+def test_batch_scan_level_filter_and_value_predicate(tmp_path, native, engine,
dv, mor, skip_l0):
+ table = _table(tmp_path, engine, dv, mor)
+ _write(table, [{'id': 1, 'v': 'base'}], 100, level=1)
+ _write(table, [{'id': 1, 'v': 'updated'}, {'id': 2, 'v': 'new'}], 200)
+ expected = ([{'id': 1, 'v': 'base'}] if skip_l0 else
+ [{'id': 1, 'v': 'updated'}, {'id': 2, 'v': 'new'}])
+ plan, rows = _read(table, native)
+ assert rows == expected
+ assert plan.snapshot_id == 2
+ assert sum(len(split.files) for split in plan.splits()) == (1 if skip_l0
else 2)
+ pb = table.new_read_builder().new_predicate_builder()
+ for value in ('base', 'updated', 'absent'):
+ _, rows = _read(table, native, pb.equal('v', value))
+ assert rows == [row for row in expected if row['v'] == value]
+ assert _read(table.copy({'scan.snapshot-id': '1'}), native)[1] == [{'id':
1, 'v': 'base'}]
+
+
[email protected]('engine,dv', [('first-row', False), ('deduplicate',
True)])
+def test_write_and_incremental_scans_keep_l0(tmp_path, engine, dv):
+ table = _table(tmp_path, engine, dv,
False).copy({'scan.native-plan.enabled': 'false'})
+ _write(table, [{'id': 1, 'v': 'base'}], 100, level=1)
+ _write(table, [{'id': 1, 'v': 'updated'}, {'id': 2, 'v': 'new'}], 200)
+ scan = table.new_read_builder().new_scan()
+ assert sum(len(split.files) for split in scan.plan().splits()) == 1
+ assert sum(len(split.files) for split in scan.plan_for_write().splits())
== 2
+ assert sum(len(split.files) for split in scan.plan().splits()) == 1
+ incremental = table.copy({'incremental-between-timestamp': '100,200'})
+ assert _read(incremental, False)[1] == [{'id': 1, 'v': 'updated'}, {'id':
2, 'v': 'new'}]
+
+
[email protected]('override', [None, 'false'])
+def test_copy_can_disable_persisted_merge_on_read(tmp_path, native, override):
+ table = _table(tmp_path, 'deduplicate', True, True)
+ _write(table, [{'id': 1, 'v': 'base'}], 100, level=1)
+ _write(table, [{'id': 1, 'v': 'updated'}], 200)
+ assert _read(table, native)[1] == [{'id': 1, 'v': 'updated'}]
+ table = table.copy({'deletion-vectors.merge-on-read': override})
+ assert _read(table, native)[1] == [{'id': 1, 'v': 'base'}]
+
+
+def test_first_row_streaming_and_overwrite_see_uncompacted_files(tmp_path):
+ from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+
+ table = _table(tmp_path, 'first-row', False, False)
+ _write(table, [{'id': 1, 'v': 'before'}], 100)
+ assert _read(table, False)[1] == []
+ stream = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ plan =
stream._create_initial_plan(table.snapshot_manager().get_latest_snapshot())
+ assert sum(len(split.files) for split in plan.splits()) == 1
+ assert
table.new_read_builder().new_read().to_arrow(plan.splits()).to_pylist() == [
+ {'id': 1, 'v': 'before'}]
+
+ _write(table, [{'id': 2, 'v': 'after'}], 200, overwrite=True)
+ builder = table.new_read_builder()
+ plan = builder.new_scan().plan_for_write()
+ assert sum(len(split.files) for split in plan.splits()) == 1
+ assert builder.new_read().to_arrow(plan.splits()).to_pylist() == [{'id':
2, 'v': 'after'}]
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
index c415f20a92..bb0c03420f 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
@@ -388,6 +388,24 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
self.assertEqual([entry.file.file_name for entry in actual],
['selected.parquet', 'partition-pruned.parquet'])
+ def test_union_manifest_schema_uses_compatible_reader(self):
+ table, manager, entries = self._partitioned_manifest()
+ buffer = BytesIO()
+ # Rust Avro manifests wrap the record schema in a top-level union.
+ fastavro.writer(buffer, [MANIFEST_ENTRY_SCHEMA],
manager._to_avro_records(entries))
+ name = 'union-manifest.avro'
+ with
table.file_io.new_output_stream('{}/{}'.format(manager.manifest_path, name)) as
output:
+ output.write(buffer.getvalue())
+
+ class KeepPartition:
+ @staticmethod
+ def test(row):
+ return row.get_field(0) == 'keep'
+
+ actual = manager.read(name, early_entry_filter=lambda bucket, _:
bucket == 1,
+ partition_filter=KeepPartition())
+ self.assertEqual([entry.file.file_name for entry in actual],
['selected.parquet'])
+
def test_manifest_bucket_and_level_stats(self):
manager = self._make_manager()
entries = [self._create_manifest_entry('a', bucket=2),
diff --git a/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
new file mode 100644
index 0000000000..1fe383e773
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
@@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Real bucket growth and cross-partition updates through native planning."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_method_available
+
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_method_available('Table', 'new_write_builder'),
+ reason='Rust main writer and planner required')]
+
+
[email protected]('cross_partition', [False, True], ids=['dynamic',
'cross-partition'])
+def test_native_pk_bucket_growth_and_partition_migration(tmp_path,
cross_partition):
+ from pypaimon_rust.datafusion import PaimonCatalog
+
+ schema = pa.schema([('id', pa.int64()), ('p', pa.string()), ('v',
pa.string())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if cross_partition else ['id', 'p'],
+ partition_keys=['p'], options={
+ 'bucket': '-1', 'dynamic-bucket.target-row-num': '2',
+ 'source.split.target-size': '1 b', 'source.split.open-file-cost':
'1 b',
+ }), False)
+
+ def read(table, native, predicate=None, shard=None, limit=None):
+ builder = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if limit is not None:
+ builder.with_limit(limit)
+ scan = builder.new_scan()
+ if shard is not None:
+ scan.with_shard(*shard)
+ if native:
+ with patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('native fallback')):
+ plan = scan.plan()
+ else:
+ plan = scan.plan()
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+ return plan, sorted(rows, key=lambda row: (row['id'], row['p']))
+
+ initial = [{'id': 1, 'p': 'a', 'v': 'old'}, {'id': 2, 'p': 'a', 'v':
'two'},
+ {'id': 3, 'p': 'b', 'v': 'three'}]
+ updates = [{'id': 1, 'p': 'b', 'v': 'moved'}, {'id': 2, 'p': 'a', 'v':
'updated'},
+ {'id': 4, 'p': 'a', 'v': 'four'}]
+ for snapshot_id, rows in enumerate((initial, updates), 1):
+ # The Rust writer maintains the dynamic/cross-partition hash index and
+ # emits the old-partition DELETE when a key moves.
+ table = PaimonCatalog({'warehouse':
str(tmp_path)}).get_table('default.t')
+ builder = table.new_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.RecordBatch.from_pylist(rows, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ table = catalog.get_table('default.t')
+ 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)):
+ python_plan, python_rows = read(table, False, predicate, shard)
+ native_plan, native_rows = read(table, True, predicate, shard)
+ assert native_rows == python_rows
+ assert native_plan.snapshot_id == python_plan.snapshot_id ==
snapshot_id
+ for native in (False, True):
+ _, limited = read(table, native, predicate, shard, limit=1)
+ assert len(limited) <= 1
+ assert all(row in python_rows for row in limited)
+
+ expected = [updates[0], updates[1], initial[2], updates[2]]
+ if not cross_partition:
+ expected.insert(0, initial[0])
+ for native in (False, True):
+ plan, actual = read(table, native)
+ assert actual == expected
+ if not cross_partition:
+ assert 1 in {split.bucket for split in plan.splits()}
+ # A time-travel scan must not use the latest migration/deletion state.
+ old_plan, old_rows = read(table.copy({'scan.snapshot-id': '1'}),
native)
+ assert old_plan.snapshot_id == 1
+ assert old_rows == initial
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index f275325eb2..8f1dcd257c 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -122,13 +122,14 @@ class NativePlanIntegrationTest(unittest.TestCase):
{'scan.snapshot-id': None, 'scan.native-plan.enabled': 'true'})
self.assertFalse(native.new_read_builder().explain().native_planned)
- def test_first_row_merge_engine_falls_back(self):
+ def test_first_row_batch_scan_uses_native_plan(self):
self.cat.create_table('default.fr_t', Schema.from_pyarrow_schema(
self.schema, primary_keys=['k'],
options={'bucket': '1', 'merge-engine': 'first-row'}), False)
self._write('fr_t', [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}])
self._write('fr_t', [{'k': 1, 'v': 'X'}, {'k': 3, 'v': 'c'}]) # k=1
stays 'a'
- self._assert_matches('fr_t', expect_native=False)
+ self._assert_matches('fr_t')
+ self.assertEqual(self._plan_and_read('fr_t', native=False)[1], [])
def test_append_matches_normal_plan(self):
self.cat.create_table(
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 6d6875163a..b61b0c8fc6 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -72,6 +72,7 @@ def _scan(native_enabled, file_scanner):
file_scanner._row_ranges = None # no explicit row ranges
file_scanner.deletion_vectors_enabled = False # no deletion vectors
file_scanner.data_evolution = False # no data evolution
+ file_scanner.skip_level0 = False
file_scanner.only_read_real_buckets = False # not postpone bucket
scan.file_scanner = file_scanner
scan.predicate = None
@@ -285,10 +286,6 @@ class NativePlanTest(unittest.TestCase):
check(lambda s, fs: setattr(fs, 'only_read_real_buckets', True))
check(lambda s, fs: (setattr(s.table, 'is_primary_key_table', True),
setattr(s.table, 'trimmed_primary_keys', [])))
- check(lambda s, fs: s.table.bucket_mode.__setattr__(
- 'return_value', BucketMode.HASH_DYNAMIC))
- check(lambda s, fs: s.table.bucket_mode.__setattr__(
- 'return_value', BucketMode.CROSS_PARTITION))
check(lambda s, fs: setattr(
s.table, '_applied_dynamic_options', {'scan.snapshot-id': None}))
check(lambda s, fs:
setattr(s.table.schema_manager.latest.return_value, 'id', 2))
@@ -581,6 +578,7 @@ class NativePlanTest(unittest.TestCase):
self.assertEqual(_read_options(table), {
'source.split.target-size': '1024',
'source.split.open-file-cost': '128',
+ 'deletion-vectors.merge-on-read': 'false',
'scan.snapshot-id': '9',
'global-index.search-mode': 'detail',
'scalar-index.search-mode': 'full',
@@ -692,7 +690,7 @@ class NativePlanTest(unittest.TestCase):
table.partition_keys = []
table.options.source_split_target_size.return_value = 1024
table.options.source_split_open_file_cost.return_value = 128
- table.options.options.contains_key.return_value = False
+ table.options.options = Options({})
table._applied_dynamic_options = {}
split = Mock()
split.serialize.return_value = b'bytes'
@@ -722,6 +720,7 @@ class NativePlanTest(unittest.TestCase):
rt.new_read_builder.assert_called_once_with({
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): '1024',
CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): '128',
+ CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(): 'false',
})
builder.with_row_ranges.assert_called_once_with([(1, 2)])
des.assert_called_once_with(b'bytes', [], kfields)
@@ -846,7 +845,7 @@ class NativePlanTest(unittest.TestCase):
'pypaimon.read.native_plan.native_plan',
return_value=Plan([], 1)) as native:
result = scan.plan()
- if version in ('0.4.0', '0.4.1') and not merge_on_read:
+ if version in ('0.4.0', '0.4.1'):
self.assertEqual(result.snapshot_id, 1)
fs.scan.assert_not_called()
else:
diff --git a/paimon-python/pypaimon/tests/schema_evolution_pk_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_pk_read_test.py
index 6fb0afd21d..675a34d727 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_pk_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_pk_read_test.py
@@ -242,8 +242,13 @@ class SchemaEvolutionPkReadTest(unittest.TestCase):
self._write(table, pa.Table.from_pydict(
{'id': [1], 'v': ['second'], 'w': ['W']}, schema=s1))
- # first-row keeps the earliest row; it predates column w, so w is NULL.
- self.assertEqual(self._read_sorted(table), [
+ # Batch first-row scans hide L0, as in Java. Inspect the un-compacted
+ # files explicitly to verify schema evolution in the merge reader.
+ builder = table.new_read_builder()
+ scan = builder.new_scan()
+ self.assertEqual(scan.plan().splits(), [])
+ rows =
builder.new_read().to_arrow(scan.plan_for_write().splits()).to_pylist()
+ self.assertEqual(rows, [
{'id': 1, 'v': 'first', 'w': None}])
# -- B8: multi-version chain (add + promotion) + partial-update ------
diff --git a/paimon-python/pypaimon/tests/table_query_auth_test.py
b/paimon-python/pypaimon/tests/table_query_auth_test.py
index ff9f74aa6a..0f28ace877 100644
--- a/paimon-python/pypaimon/tests/table_query_auth_test.py
+++ b/paimon-python/pypaimon/tests/table_query_auth_test.py
@@ -91,6 +91,7 @@ class _FakeScan:
class _FakeFileScanner:
def __init__(self, plan):
self._plan = plan
+ self.skip_level0 = False
def scan(self):
return self._plan
diff --git a/paimon-python/pypaimon/tests/test_first_row_e2e.py
b/paimon-python/pypaimon/tests/test_first_row_e2e.py
index 58776848b8..5722ad973c 100644
--- a/paimon-python/pypaimon/tests/test_first_row_e2e.py
+++ b/paimon-python/pypaimon/tests/test_first_row_e2e.py
@@ -19,8 +19,9 @@
"""End-to-end tests for the ``first-row`` merge engine.
Each test creates a PK table with ``merge-engine`` set to ``first-row``,
-writes one or more batches, and reads back. The first-row engine keeps
-only the earliest row per primary key.
+writes one or more batches, and verifies the un-compacted files explicitly.
+Java batch scans hide first-row L0; the internal write scan exposes them for
+checking write-buffer folding and the first-row merge function.
"""
import os
@@ -81,7 +82,9 @@ class FirstRowMergeEngineE2ETest(unittest.TestCase):
def _read(self, table):
rb = table.new_read_builder()
- splits = rb.new_scan().plan().splits()
+ scan = rb.new_scan()
+ self.assertEqual(scan.plan().splits(), [])
+ splits = scan.plan_for_write().splits()
if not splits:
return []
return sorted(
diff --git a/paimon-python/pypaimon/tests/test_partial_update_e2e.py
b/paimon-python/pypaimon/tests/test_partial_update_e2e.py
index 90fa5f711b..f0a15b1fcb 100644
--- a/paimon-python/pypaimon/tests/test_partial_update_e2e.py
+++ b/paimon-python/pypaimon/tests/test_partial_update_e2e.py
@@ -282,8 +282,10 @@ class PartialUpdateMergeEngineE2ETest(unittest.TestCase):
self._write(table, [{'id': 1, 'a': 'first', 'b': None, 'c': None}])
self._write(table, [{'id': 1, 'a': 'second', 'b': 'B', 'c': 'C'}])
+ self.assertEqual(self._read(table), [])
+ builder = table.new_read_builder()
self.assertEqual(
- self._read(table),
+
builder.new_read().to_arrow(builder.new_scan().plan_for_write().splits()).to_pylist(),
[{'id': 1, 'a': 'first', 'b': None, 'c': None}],
)
