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 8790fdca94b6e15f2e05ef369d97f2434e8c5f8c 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}], )
