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 8a429cfc363149b7d7078603c90fa8e730b664db Author: Jingsong Lee <[email protected]> AuthorDate: Wed Sep 16 13:07:08 2026 +0800 [python] Expand native planning across resolved tables and scan modes (#9873) --- paimon-python/README.md | 35 +++- paimon-python/pypaimon/common/json_util.py | 2 + .../apply_deletion_vector_reader.py | 13 +- paimon-python/pypaimon/read/native_plan.py | 84 +++++++++- .../pypaimon/read/reader/format_pyarrow_reader.py | 3 + .../read/scanner/chunk_shuffle_split_generator.py | 13 +- paimon-python/pypaimon/read/split_read.py | 18 +- paimon-python/pypaimon/read/table_scan.py | 151 +++++++++++------ paimon-python/pypaimon/table/file_store_table.py | 5 +- .../tests/native_plan_chunk_shuffle_test.py | 172 +++++++++++++++++++ .../pypaimon/tests/native_plan_integration_test.py | 13 +- .../tests/native_plan_materialized_pk_test.py | 186 +++++++++++++++++++++ .../tests/native_plan_resolved_schema_test.py | 184 ++++++++++++++++++++ .../pypaimon/tests/native_plan_rest_test.py | 151 +++++++++++++++++ paimon-python/pypaimon/tests/native_plan_test.py | 57 ++++++- .../pypaimon/tests/reader_predicate_test.py | 78 +++++++++ .../pypaimon/tests/resolving_file_io_test.py | 21 +++ .../pypaimon/tests/schema_evolution_read_test.py | 53 +++++- .../pypaimon/tests/snapshot_manager_test.py | 24 +++ 19 files changed, 1177 insertions(+), 86 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index 98dc4fc082..2af01d988c 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -50,6 +50,23 @@ Python planner for unsupported scans. New bindings preserve `plan.snapshot_id` even when pruning removes every split. Native explain output includes snapshot and split metadata; native pruning counters are not exposed. +With Rust main's `Table.from_resolved_schema()` binding, filesystem and JDBC catalog +tables preserve the Python table's resolved schema and complete effective +options. Stale table objects, historical schemas, and `copy()` overrides or +option removals no longer require catalog reloading or Python planning. +Tables opened with `FileStoreTable.from_path(path, file_io_options=None)` use +the same path with standard local, PyArrow or resolving FileIO. Storage options +configure FileIO; use `copy()` for table read options. +JDBC planning uses the resolved table location and storage properties without +opening another database connection. +REST tables use `Table.copy_with_resolved_schema()` to preserve the same schema +and option semantics, including branches whose schemas are catalog-managed. +The native table retains REST credentials, token refresh and catalog snapshot +resolution. Database and table names containing dots are passed as separate +identifier components. REST snapshot results (including empty results) take precedence over +filesystem snapshots. REST errors, including HTTP 501, are propagated as in Java. +Custom catalog/FileIO contexts still fall back when they cannot be reproduced. + Explicit row ranges on data-evolution tables require `ReadBuilder.with_row_ranges()`. Watermark time travel requires Rust 0.4 or newer. Branch reads require the branch-aware binding exposing `Table.branch()`, and the resolved branch is @@ -90,14 +107,28 @@ bucket sharding. Cross-partition key migration is maintained by the writer's ind 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. +together when they include L0 and require reader-side merging. Fully materialized +DV files across levels use raw splits, including first-row clustering tables. +First-row L0 runs can use native planning, including plans with materialized files +in separate raw splits. Plans that require merging clustered materialized files +still fall back to Python. Readers preserve physical row positions until deletion +vectors are applied, then evaluate residual predicates after merging. +Write scans and incremental scans retain level 0. + +Append and data-evolution chunk shuffle use Rust file and deletion-vector planning. +Python retains live-row chunk sizing, seeded shuffle order and balanced worker +assignment, so the same seed selects the same chunks with either planner. +Projection does not remove aligned column files before chunk construction. +Chunk shuffle supports partition predicates, deletion vectors and timestamp +incremental scans; its existing restrictions on limits, slices, row ranges and +global-index results still apply. Scored global-index results on data-evolution append tables use native row-range planning; Python attaches scores to the selected ranges and reads the data. Primary-key sorted indexes refine native batch splits through Python's existing index reader, preserving merge-required splits and the selected snapshot. -Chunk shuffle, query authorization, batch first-row scans explicitly including L0, +Query authorization, first-row plans mixing L0 with merge-required materialized files, and precomputed primary-key global-index results still use the Python planner. Continuous streaming and write planning also retain their Python entrypoints. Native planning remains optional and is disabled by default. diff --git a/paimon-python/pypaimon/common/json_util.py b/paimon-python/pypaimon/common/json_util.py index 6effec7d1e..0ae8c232f9 100644 --- a/paimon-python/pypaimon/common/json_util.py +++ b/paimon-python/pypaimon/common/json_util.py @@ -148,6 +148,8 @@ class JSON: kwargs[field_name] = ( None if value is None else decoder_mapping[json_name](value) ) + elif value is None: + kwargs[field_name] = None elif json_name in type_mapping: tp = getattr(type_mapping[json_name], '__origin__', None) if tp in (list, List): diff --git a/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py b/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py index f491ecfa3f..237dae3775 100644 --- a/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py +++ b/paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py @@ -75,6 +75,7 @@ class ApplyDeletionVectorReader(RecordBatchReader): self._reader = reader self._deletion_vector = deletion_vector self._returned_position = 0 + self._record_iterator = None def reader(self) -> RecordReader: return self._reader @@ -108,12 +109,18 @@ class ApplyDeletionVectorReader(RecordBatchReader): Returns: A RecordIterator with deletion filtering, or None if no more data. """ + if self._record_iterator is not None: + self._returned_position = self._record_iterator.returned_position() + 1 batch = self._reader.read_batch() if batch is None: return None - return ApplyDeletionRecordIterator(batch, self._deletion_vector) + # Positions address the whole file, including rows deleted in earlier + # batches. Starting each iterator at zero would apply the DV repeatedly. + self._record_iterator = ApplyDeletionRecordIterator( + batch, self._deletion_vector, self._returned_position) + return self._record_iterator def close(self): self._reader.close() @@ -129,6 +136,7 @@ class ApplyDeletionRecordIterator(RecordIterator): self, iterator: RecordIterator, deletion_vector, + start_position: int = 0, ): """ Initialize an ApplyDeletionRecordIterator. @@ -136,10 +144,11 @@ class ApplyDeletionRecordIterator(RecordIterator): Args: iterator: The underlying record iterator. deletion_vector: The deletion vector to apply for filtering. + start_position: Position of the first record in this batch. """ self._iterator = iterator self._deletion_vector = deletion_vector - self._returned_position = -1 + self._returned_position = start_position - 1 def iterator(self) -> RecordIterator: return self._iterator diff --git a/paimon-python/pypaimon/read/native_plan.py b/paimon-python/pypaimon/read/native_plan.py index 253eeabb8f..ed12a98948 100644 --- a/paimon-python/pypaimon/read/native_plan.py +++ b/paimon-python/pypaimon/read/native_plan.py @@ -114,16 +114,23 @@ def _catalog_options(table) -> dict: loader = getattr(getattr(table, 'catalog_environment', None), 'catalog_loader', None) if loader is None: raise ValueError("native_plan requires a catalog-backed table (no catalog loader)") + metastore = _catalog_metastore(loader) + if metastore is None: + raise ValueError("native_plan requires an exact built-in catalog loader") + normalized = _catalog_context_options(table) + normalized[CatalogOptions.METASTORE.key()] = metastore + return normalized + + +def _catalog_context_options(table) -> dict: + """Normalize catalog storage properties without rebuilding the metastore.""" + loader = table.catalog_environment.catalog_loader options = loader.context().options.to_map() normalized = { str(key): _option_value_to_string(value) for key, value in options.items() if value is not None } - metastore = _catalog_metastore(loader) - if metastore is None: - raise ValueError("native_plan requires an exact built-in catalog loader") - normalized[CatalogOptions.METASTORE.key()] = metastore if str(getattr(table, 'table_path', '')).startswith('oss://'): from pypaimon.filesystem.jindo_file_system_handler import ( JINDO_AVAILABLE, @@ -169,6 +176,47 @@ def _read_options(table) -> dict: return options +def _resolved_schema_file_io_options(table) -> Optional[dict]: + """FileIO properties for tables whose metadata needs no catalog resolution.""" + if not native_method_available('Table', 'from_resolved_schema'): + return None + environment = table.catalog_environment + loader = environment.catalog_loader + if loader is None: + from pypaimon.catalog.catalog_environment import CatalogEnvironment + from pypaimon.filesystem.local_file_io import LocalFileIO + from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO + from pypaimon.filesystem.resolving_file_io import ResolvingFileIO + # A custom environment or FileIO can supply metadata outside the path. + if (type(environment) is not CatalogEnvironment + or type(table.file_io) not in (LocalFileIO, PyArrowFileIO, ResolvingFileIO)): + return None + return {str(key): _option_value_to_string(value) + for key, value in table.file_io.properties.to_map().items() + if value is not None} + from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader + if _catalog_metastore(loader) != 'filesystem' and type(loader) is not JdbcCatalogLoader: + # REST tables must retain catalog snapshot loading and token refresh. + return None + context = loader.context() + if context.options is None or any(getattr(context, attr, None) is not None for attr in ( + 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')): + return None + # JDBC, like filesystem catalogs, uses on-disk snapshots. Its already + # resolved table does not need another database connection during planning. + return _catalog_context_options(table) + + +def _resolved_schema_json(table) -> str: + from pypaimon.common.json_util import JSON + options = {str(key): _option_value_to_string(value) + for key, value in table.table_schema.options.items() if value is not None} + options.update(_read_options(table)) + # The timestamp string has already been converted to epoch millis. + options.pop(CoreOptions.SCAN_TIMESTAMP.key(), None) + return JSON.to_json(table.table_schema.copy(new_options=options)) + + def _predicate_to_native(predicate: Predicate) -> dict: """Convert PyPaimon's predicate tree to pypaimon-rust's dict API.""" if predicate.method in ('and', 'or'): @@ -231,14 +279,34 @@ def native_plan( if not native_runtime_available(): raise RuntimeError( "scan.native-plan.enabled needs pypaimon-rust>=0.3.0 (split planning API)") - from pypaimon_rust.datafusion import PaimonCatalog - - rt = PaimonCatalog(_catalog_options(table)).get_table(table.identifier.get_full_name()) + file_io_options = _resolved_schema_file_io_options(table) + if file_io_options is not None: + from pypaimon_rust.datafusion import Table + rt = Table.from_resolved_schema( + table.table_path, _resolved_schema_json(table), + database=table.identifier.get_database_name(), + table=table.identifier.get_table_name(), + branch=table.current_branch(), options=file_io_options) + builder = rt.new_read_builder() + else: + from pypaimon_rust.datafusion import PaimonCatalog + catalog = PaimonCatalog(_catalog_options(table)) + if native_method_available('Table', 'copy_with_resolved_schema'): + # REST may keep branch schemas in the catalog only. Load the base + # environment, then attach the schema/branch already resolved here. + rt = catalog.get_table(( + table.identifier.get_database_name(), table.identifier.get_table_name())) + if rt.location() != table.table_path: + raise RuntimeError('Native catalog resolved a different table location') + rt = rt.copy_with_resolved_schema(_resolved_schema_json(table), branch=table.current_branch()) + builder = rt.new_read_builder() + else: + rt = catalog.get_table(table.identifier.get_full_name()) + builder = rt.new_read_builder(_read_options(table)) if table.current_branch() != 'main': branch = getattr(rt, 'branch', None) if not callable(branch) or branch() != table.current_branch(): raise RuntimeError("Native table did not resolve the requested branch") - builder = rt.new_read_builder(_read_options(table)) if projection is not None: builder = builder.with_projection(projection) if predicate is not None: diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py index 7ce7ec1d7a..33999fec75 100644 --- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py +++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py @@ -341,6 +341,9 @@ class FormatPyArrowReader(RecordBatchReader): row_indices: Optional[List[int]] = None, row_ranges: Optional[List[Tuple[int, int]]] = None, row_group_cache: Optional[_DecodedRowGroupCache] = None): + from pypaimon.filesystem.resolving_file_io import ResolvingFileIO + if isinstance(file_io, ResolvingFileIO): + file_io = file_io._get_fileio(file_path) self._predicate_field_names = predicate_field_names or set() file_path_for_pyarrow = file_io.to_filesystem_path(file_path) self._row_group_cache = row_group_cache diff --git a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py index 89216322cf..33881bafcb 100644 --- a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py +++ b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py @@ -230,12 +230,13 @@ class ChunkShuffleSplitGeneratorBase(AbstractSplitGenerator): if f.file_name in seen_paths: continue seen_paths.add(f.file_name) - f.set_file_path( - self.table.table_path, - partition_row, - bucket, - self.default_part_value, - ) + if not f.file_path: + f.set_file_path( + self.table.table_path, + partition_row, + bucket, + self.default_part_value, + ) for segments in self._slice_group_into_chunks(entries_in_group): all_chunks.append(_Chunk(partition_row, bucket, segments)) diff --git a/paimon-python/pypaimon/read/split_read.py b/paimon-python/pypaimon/read/split_read.py index ec1ee56689..4f1fbd64bf 100644 --- a/paimon-python/pypaimon/read/split_read.py +++ b/paimon-python/pypaimon/read/split_read.py @@ -240,6 +240,13 @@ class SplitRead(ABC): read_arrow_predicate, read_paimon_predicate, ) = self._get_fields_and_predicate(file.schema_id, read_fields) + if (file.file_name in self.deletion_file_readers + or (for_merge_read and self.row_ranges is not None)): + # DVs and indexed PK ranges refer to physical file positions. + # Filtering or skipping row groups here would renumber those rows. + # Apply the residual predicate after position selection and merging. + read_arrow_predicate = None + read_paimon_predicate = None # Use external_path if available, otherwise use file_path file_path = file.external_path if file.external_path else file.file_path @@ -583,7 +590,11 @@ class SplitRead(ABC): read_field for read_field in read_fields if _is_reachable(read_field) ] - read_predicate = trim_predicate_by_fields(self.push_down_predicate, read_file_fields) + # File readers filter physical column names before field-id schema + # normalization. A renamed or re-added name can identify a different + # column, so cross-schema filtering must run after normalization. + read_predicate = (trim_predicate_by_fields(self.push_down_predicate, read_file_fields) + if schema_id == self.table.table_schema.id else None) read_arrow_predicate = ( read_predicate.to_arrow() if read_predicate and self._arrow_filter_pushdown_enabled @@ -899,7 +910,10 @@ class RawFileSplitRead(SplitRead): reader = concat_reader if (self.predicate_for_reader and (self.table.is_primary_key_table - or not self._arrow_filter_pushdown_enabled)): + or not self._arrow_filter_pushdown_enabled + or self.deletion_file_readers + or any(file.schema_id != self.table.table_schema.id + for file in self.split.files))): reader = FilterRecordBatchReader( reader, self.predicate_for_reader, diff --git a/paimon-python/pypaimon/read/table_scan.py b/paimon-python/pypaimon/read/table_scan.py index 7e4889da51..74a5be1e6b 100755 --- a/paimon-python/pypaimon/read/table_scan.py +++ b/paimon-python/pypaimon/read/table_scan.py @@ -115,25 +115,26 @@ class TableScan: def _native_plan_supported_impl(self) -> bool: """Fall back to the Python scanner for scans native can't carry: - chunk-shuffle, primary-key global-index results, first-row scans which include L0, + primary-key global-index results, a primary-key table whose trimmed PK is empty (PK equals the partition - key; Rust rejects this schema), a stale - schema without time travel, removed copy() options which Rust cannot - represent, unsupported time travel selectors, + key; Rust rejects this schema), unsupported time travel selectors, + and catalog-loaded tables with schema/option overrides Rust cannot carry, query auth, a missing/old pypaimon-rust, or a catalog / identifier Rust cannot reconstruct. Keep this capability gate in sync when adding scan features.""" from pypaimon.read.native_plan import ( - native_method_available, native_runtime_available, native_version_at_least, + _resolved_schema_file_io_options, native_method_available, + native_runtime_available, native_version_at_least, ) if not native_runtime_available(): return False fs = self.file_scanner if fs.is_streaming and not native_method_available('Split', 'is_streaming'): return False - if (getattr(fs, 'chunk_shuffle', None) is not None - or not self._native_global_index_result_supported()): + if not self._native_global_index_result_supported(): return False + if getattr(fs, 'chunk_shuffle', None) is not None: + fs._validate_chunk_shuffle_compat() # Positional append distribution needs the stable partition/file order # introduced in 0.4. Older bindings can assign different rows per call. if (not self.table.is_primary_key_table and not fs.data_evolution @@ -157,35 +158,34 @@ class TableScan: if (self.table.current_branch() != 'main' and not native_method_available('Table', 'branch')): return False - loader = getattr( - getattr(self.table, 'catalog_environment', None), - 'catalog_loader', - None, - ) - context_fn = getattr(loader, 'context', None) - if not callable(context_fn): - return False - from pypaimon.read.native_plan import _catalog_metastore - if _catalog_metastore(loader) is None: - return False - context = context_fn() - catalog_options = getattr(context, 'options', None) - if catalog_options is None: - return False - if any(getattr(context, attr, None) is not None for attr in ( - 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')): - return False - database_name = self.table.identifier.get_database_name() - if not database_name or database_name == UNKNOWN_DATABASE or '.' in database_name: - return False + resolved_schema = _resolved_schema_file_io_options(self.table) is not None + if not resolved_schema: + loader = getattr( + getattr(self.table, 'catalog_environment', None), + 'catalog_loader', + None, + ) + context_fn = getattr(loader, 'context', None) + if not callable(context_fn): + return False + from pypaimon.read.native_plan import _catalog_metastore + if _catalog_metastore(loader) is None: + return False + context = context_fn() + catalog_options = getattr(context, 'options', None) + if catalog_options is None: + return False + if any(getattr(context, attr, None) is not None for attr in ( + 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')): + return False + database_name = self.table.identifier.get_database_name() + if not database_name or database_name == UNKNOWN_DATABASE: + return False + if ('.' in database_name + and not native_method_available('Table', 'copy_with_resolved_schema')): + return False if self.table.options.query_auth_enabled: return False - # 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 # 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: @@ -199,23 +199,24 @@ class TableScan: from pypaimon.read.native_plan import native_family_search_modes_available if not native_family_search_modes_available(): return False - supported_time_travel = any( - options.contains_key(key) for key in _NATIVE_TIME_TRAVEL_OPTIONS) - # Time travel intentionally carries a historical schema; other stale - # table objects must still fall back because Rust reloads the latest. - latest_schema = self.table.schema_manager.latest() - if (not supported_time_travel and latest_schema is not None - and latest_schema.id != self.table.table_schema.id): - return False - # Rust cannot remove an option persisted in the catalog-loaded schema. - applied_options = getattr(self.table, '_applied_dynamic_options', {}) or {} - allowed_options = ( - _NATIVE_FORWARDED_OPTIONS | _NATIVE_PLAN_INDEPENDENT_OPTIONS) - if (set(applied_options) - allowed_options - or any(key in (_NATIVE_TIME_TRAVEL_OPTIONS - | _NATIVE_SEARCH_MODE_OPTIONS) and value is None - for key, value in applied_options.items())): - return False + if not resolved_schema and not native_method_available('Table', 'copy_with_resolved_schema'): + supported_time_travel = any( + options.contains_key(key) for key in _NATIVE_TIME_TRAVEL_OPTIONS) + # Time travel intentionally carries a historical schema; other stale + # table objects must still fall back because Rust reloads the latest. + latest_schema = self.table.schema_manager.latest() + if (not supported_time_travel and latest_schema is not None + and latest_schema.id != self.table.table_schema.id): + return False + # Rust cannot remove an option persisted in the catalog-loaded schema. + applied_options = getattr(self.table, '_applied_dynamic_options', {}) or {} + allowed_options = ( + _NATIVE_FORWARDED_OPTIONS | _NATIVE_PLAN_INDEPENDENT_OPTIONS) + if (set(applied_options) - allowed_options + or any(key in (_NATIVE_TIME_TRAVEL_OPTIONS + | _NATIVE_SEARCH_MODE_OPTIONS) and value is None + for key, value in applied_options.items())): + return False from pypaimon.snapshot.time_travel_util import SCAN_KEYS unsupported_scan_keys = set(SCAN_KEYS) - _NATIVE_TIME_TRAVEL_OPTIONS if any(options.contains_key(k) for k in unsupported_scan_keys): @@ -261,7 +262,8 @@ class TableScan: if self._incremental_snapshot_range is None: return Plan([]) extra_options['incremental_range'] = self._incremental_snapshot_range - if has_distribution and fs.data_evolution: + chunk_shuffle = fs.chunk_shuffle + if has_distribution and fs.data_evolution and chunk_shuffle is None: if fs.idx_of_this_subtask is not None: extra_options['row_position_shard'] = ( fs.idx_of_this_subtask, fs.number_of_para_subtasks) @@ -283,11 +285,20 @@ class TableScan: limit=None if has_distribution else self.limit, projection=( [field.name for field in self._read_type] - if self._read_type is not None else None), + if self._read_type is not None and chunk_shuffle is None else None), row_ranges=row_ranges, **extra_options, ) splits = plan.splits() + if (self.table.options.merge_engine() == 'first-row' + and not fs.skip_level0 and not fs.is_streaming + and any(file.level == 0 for split in splits for file in split.files) + and any(not split.raw_convertible and any(file.level > 0 for file in split.files) + for split in splits)): + # L0 runs are sorted by PK and can merge using first-row. + # Materialized clustered files must stay in raw splits: their + # physical order need not match the PK merge comparator. + return None partition_predicate = self.file_scanner.partition_key_predicate if partition_predicate is not None: splits = [s for s in splits @@ -298,7 +309,9 @@ class TableScan: and plan.snapshot_id is not None): snapshot = self.table.snapshot_manager().get_snapshot_by_id(plan.snapshot_id) splits = fs._apply_primary_key_sorted_indexes(splits, snapshot) - if has_distribution: + if chunk_shuffle is not None: + splits = self._chunk_shuffle_splits(splits, plan.snapshot_id) + elif has_distribution: if self.table.is_primary_key_table: splits = [s for s in splits if s.bucket % fs.number_of_para_subtasks == fs.idx_of_this_subtask] @@ -328,6 +341,36 @@ class TableScan: "Native plan failed, falling back to the Python scanner: %s", e) return None + def _chunk_shuffle_splits(self, splits, snapshot_id): + """Reuse native file/DV planning with Python's stable chunk assignment.""" + from pypaimon.manifest.schema.manifest_entry import ManifestEntry + from pypaimon.read.scanner.chunk_shuffle_split_generator import ( + AppendChunkShuffleSplitGenerator, DataEvolutionChunkShuffleSplitGenerator, + ) + fs = self.file_scanner + entries, deletions = [], {} + for split in splits: + key = (tuple(split.partition.values), split.bucket) + for index, file in enumerate(split.files): + entries.append(ManifestEntry( + 0, split.partition, split.bucket, self.table.total_buckets, file)) + if split.data_deletion_files and split.data_deletion_files[index] is not None: + deletions.setdefault(key, {})[file.file_name] = split.data_deletion_files[index] + generator_type = (DataEvolutionChunkShuffleSplitGenerator if fs.data_evolution + else AppendChunkShuffleSplitGenerator) + seed, chunk_size = fs.chunk_shuffle + generator = generator_type(self.table, fs.target_split_size, fs.open_file_cost, + deletions, seed=seed, chunk_size=chunk_size) + if fs.idx_of_this_subtask is not None: + generator.with_shard(fs.idx_of_this_subtask, fs.number_of_para_subtasks) + chunks = generator.create_splits(entries) + for split in chunks: + while callable(getattr(split, 'data_split', None)): + split = split.data_split() + split.snapshot_id = snapshot_id + split.is_streaming = fs.is_streaming + return chunks + def plan_for_write(self) -> Plan: if self.__auth_query() is not None: raise TableNoPermissionException(self.table.identifier) diff --git a/paimon-python/pypaimon/table/file_store_table.py b/paimon-python/pypaimon/table/file_store_table.py index 5519ee087d..c2962236b9 100644 --- a/paimon-python/pypaimon/table/file_store_table.py +++ b/paimon-python/pypaimon/table/file_store_table.py @@ -64,12 +64,13 @@ class FileStoreTable(Table): file_io, table_path, branch=self.current_branch()) @classmethod - def from_path(cls, table_path: str) -> 'FileStoreTable': + def from_path(cls, table_path: str, file_io_options: Optional[dict] = None) -> 'FileStoreTable': """ Create a FileStoreTable from a table path. This is useful for reading tables created by Java without going through a catalog. + ``file_io_options`` configures storage access; use ``copy`` for table read options. """ - file_io = FileIO(table_path, Options({})) + file_io = FileIO.get(table_path, Options(file_io_options or {})) schema_manager = SchemaManager(file_io, table_path) table_schema = schema_manager.latest() diff --git a/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py b/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py new file mode 100644 index 0000000000..c2755aca17 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_plan_chunk_shuffle_test.py @@ -0,0 +1,172 @@ +# 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. + +import json +from collections import Counter +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import CatalogFactory, Schema +from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector +from pypaimon.read.native_plan import native_runtime_available +from pypaimon.write.commit_message import CommitMessage +from pypaimon.write.table_delete import TableDeleteByRowId + + +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_runtime_available(), reason='Rust planner required')] + + [email protected](params=['append', 'append-dv', 'de', 'de-dv']) +def chunk_table(request, tmp_path): + de = request.param.startswith('de') + dv = request.param.endswith('dv') + fields = [('id', pa.int64()), ('p', pa.string())] + options = {'file.format': 'parquet', 'source.split.target-size': '1b', + 'source.split.open-file-cost': '1b'} + if de: + fields.append(('payload', pa.large_binary())) + options.update({'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true', + 'blob.target-file-size': '1b'}) + if dv: + options['deletion-vectors.enabled'] = 'true' + schema = pa.schema(fields) + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('default', True) + catalog.create_table('default.t', Schema.from_pyarrow_schema( + schema, partition_keys=['p'], options=options), False) + table = catalog.get_table('default.t') + expected = [] + for start in (0, 6): + rows = [{'id': i, 'p': (None, 'a', 'b')[i % 3]} for i in range(start, start + 6)] + if de: + for row in rows: + row['payload'] = ('payload-%d' % row['id']).encode() + expected.extend(rows) + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(rows, schema=schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + for snapshot_id in (1, 2): + 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) + before_deletes = list(expected) + if dv: + builder = table.copy({'scan.native-plan.enabled': 'false'}).new_read_builder() + plan = builder.new_scan().plan() + index_adds = [] + for split in plan.splits(): + # Delete the first physical row in every main file. Sidecar files + # inherit the anchor's DV through their aligned row-id group. + for file in split.files: + if not file.file_name.endswith('.parquet'): + continue + from pypaimon.read.split import DataSplit + one = DataSplit(files=[file], partition=split.partition, bucket=split.bucket, + raw_convertible=not de) + row = builder.with_projection(['id']).new_read().to_arrow([one]).to_pylist()[0] + expected = [candidate for candidate in expected if candidate['id'] != row['id']] + vector = BitmapDeletionVector() + vector.delete(0) + index_adds.append((tuple(split.partition.values), split.bucket, + TableDeleteByRowId(table)._write_deletion_vector_index( + split.partition, split.bucket, {file.file_name: vector}))) + commit = table.new_batch_write_builder().new_commit() + try: + commit.commit([CommitMessage(partition=partition, bucket=bucket, new_files=[], + index_adds=[entry]) for partition, bucket, entry in index_adds]) + finally: + commit.close() + return table, expected, before_deletes + + +def _chunks(table, seed, chunk_size=3, shard=None, predicate=None, projection=None): + results = [] + for native in (False, True): + builder = table.copy({'scan.native-plan.enabled': str(native).lower()}).new_read_builder() + if predicate is not None: + builder.with_filter(predicate) + if projection is not None: + builder.with_projection(projection) + scan = builder.new_scan().with_chunk_shuffle(seed, chunk_size) + if shard is not None: + scan.with_shard(*shard) + if native: + with patch.object(scan.file_scanner, 'scan', side_effect=AssertionError('native fallback')), \ + patch.object(scan.file_scanner, 'plan_files', side_effect=AssertionError('Python manifest scan')): + plan = scan.plan() + else: + plan = scan.plan() + chunks = [] + for split in plan.splits(): + if table.options.options.contains_key('incremental-between-timestamp'): + assert split.is_streaming + rows = builder.new_read().to_arrow([split]).to_pylist() + assert 0 < len(rows) <= chunk_size + assert split.merged_row_count() == len(rows) + if rows and 'p' in rows[0]: + assert len({row['p'] for row in rows}) == 1 + chunks.append(sorted(rows, key=lambda row: row['id'])) + results.append((plan.snapshot_id, chunks)) + assert results[0] == results[1] + return results[1] + + [email protected]('seed', [-11, 42, 2 ** 70]) +def test_native_chunks_preserve_order_live_counts_and_worker_assignment(chunk_table, seed): + table, expected, _ = chunk_table + snapshot_id, chunks = _chunks(table, seed) + actual = [row for chunk in chunks for row in chunk] + assert sorted(actual, key=lambda row: row['id']) == expected + workers = [] + for worker in range(5): + sid, assigned = _chunks(table, seed, shard=(worker, 5)) + assert sid == snapshot_id + workers.extend(assigned) + assert workers == chunks + assert Counter(row['id'] for chunk in workers for row in chunk) == Counter(row['id'] for row in expected) + + +def test_native_chunks_keep_partition_filter_projection_and_time_travel(chunk_table): + table, expected, before_deletes = chunk_table + pb = table.new_read_builder().new_predicate_builder() + for predicate, partition in ((pb.equal('p', 'a'), 'a'), (pb.is_null('p'), None)): + _, chunks = _chunks(table, 7, predicate=predicate, projection=['id']) + assert sorted(row['id'] for chunk in chunks for row in chunk) == [ + row['id'] for row in expected if row['p'] == partition] + sid, chunks = _chunks(table.copy({'scan.snapshot-id': '2'}), 7) + assert sid == 2 + assert sorted([row for chunk in chunks for row in chunk], key=lambda row: row['id']) == before_deletes + sid, chunks = _chunks(table, 7, predicate=pb.equal('p', 'missing')) + assert sid == table.snapshot_manager().get_latest_snapshot().id + assert chunks == [] + + +def test_native_incremental_chunks_keep_events_and_ignore_later_deletions(chunk_table): + table, _, before_deletes = chunk_table + incremental = table.copy({'incremental-between-timestamp': '0,200'}) + sid, chunks = _chunks(incremental, 42) + assert sid == 2 + assert sorted([row for chunk in chunks for row in chunk], key=lambda row: row['id']) == before_deletes diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py b/paimon-python/pypaimon/tests/native_plan_integration_test.py index 8f1dcd257c..10cc5d3471 100644 --- a/paimon-python/pypaimon/tests/native_plan_integration_test.py +++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py @@ -112,15 +112,22 @@ class NativePlanIntegrationTest(unittest.TestCase): rb = native_table.new_read_builder() rb.new_read().to_arrow(rb.new_scan().plan().splits()) - def test_copy_removed_persisted_scan_option_falls_back(self): - # copy() removes a persisted scan.snapshot-id that Rust would still reload -> fall back. + def test_copy_removed_persisted_scan_option_uses_native(self): + # The resolved schema must replace, rather than merge, persisted options. self.cat.create_table('default.snapopt_t', Schema.from_pyarrow_schema( self.schema, options={'scan.snapshot-id': '1'}), False) self._write('snapopt_t', [{'k': 1, 'v': 'a'}]) # snapshot 1 self._write('snapopt_t', [{'k': 2, 'v': 'b'}]) # snapshot 2 native = self.cat.get_table('default.snapopt_t').copy( {'scan.snapshot-id': None, 'scan.native-plan.enabled': 'true'}) - self.assertFalse(native.new_read_builder().explain().native_planned) + builder = native.new_read_builder() + scan = builder.new_scan() + with patch.object(scan.file_scanner, 'scan', side_effect=AssertionError('native fallback')): + plan = scan.plan() + self.assertEqual(plan.snapshot_id, 2) + self.assertEqual(sorted(builder.new_read().to_arrow(plan.splits()).to_pylist(), + key=lambda row: row['k']), + [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}]) def test_first_row_batch_scan_uses_native_plan(self): self.cat.create_table('default.fr_t', Schema.from_pyarrow_schema( diff --git a/paimon-python/pypaimon/tests/native_plan_materialized_pk_test.py b/paimon-python/pypaimon/tests/native_plan_materialized_pk_test.py new file mode 100644 index 0000000000..6e9471088f --- /dev/null +++ b/paimon-python/pypaimon/tests/native_plan_materialized_pk_test.py @@ -0,0 +1,186 @@ +# 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. + +from dataclasses import replace +from pathlib import Path +from unittest.mock import patch + +import pyarrow as pa +import pyarrow.parquet as pq +import pytest + +from pypaimon import CatalogFactory, Schema +from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector +from pypaimon.read.native_plan import native_runtime_available +from pypaimon.table.row.generic_row import GenericRow +from pypaimon.write.commit_message import CommitMessage +from pypaimon.write.table_delete import TableDeleteByRowId + + +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_runtime_available(), reason='Rust planner required')] + + [email protected]('engine', ['deduplicate', 'first-row']) [email protected]('merge_on_read', ['false', 'true']) [email protected]('target_size', ['1b', '1mb']) +def test_clustered_materialized_dv_files_use_native_raw_splits(tmp_path, engine, merge_on_read, target_size): + schema = pa.schema([('id', pa.int64()), ('value', 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'], options={ + 'bucket': '1', 'merge-engine': engine, 'file.format': 'parquet', + 'deletion-vectors.enabled': 'true', 'deletion-vectors.merge-on-read': merge_on_read, + 'pk-clustering-override': 'true', 'clustering.columns': 'value', + 'source.split.target-size': target_size, 'source.split.open-file-cost': '1b', + }), False) + table = catalog.get_table('default.t') + expected = [{'id': i, 'value': 'v%d' % (10 - i)} for i in range(1, 5)] + files = [] + 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=schema)) + messages = writer.prepare_commit() + for message in messages: + rewritten = [] + for file in message.new_files: + path = table.path_factory().bucket_path((), 0) + '/' + file.file_name + # Simulate Java clustering compaction: physical order is by + # value, opposite to PK order, while min/max PK ranges overlap. + data = pq.read_table(path).sort_by([('value', 'ascending')]) + pq.write_table(data, path) + rewritten.append(replace(file, level=level, + file_size=Path(path).stat().st_size)) + message.new_files = rewritten + files.extend(rewritten) + commit.commit(messages) + finally: + writer.close() + commit.close() + + vector = BitmapDeletionVector() + vector.delete(0) # clustered first file starts with id=3 + entry = TableDeleteByRowId(table)._write_deletion_vector_index( + GenericRow([], []), 0, {files[0].file_name: vector}) + commit = table.new_batch_write_builder().new_commit() + try: + commit.commit([CommitMessage(partition=(), bucket=0, new_files=[], index_adds=[entry])]) + finally: + commit.close() + expected = [row for row in expected if row['id'] != 3] + pb = table.new_read_builder().new_predicate_builder() + for predicate in (None, pb.equal('id', 3), pb.equal('value', 'v8')): + for native in (False, True): + 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() + assert all(split.raw_convertible for split in plan.splits()) + rows = builder.new_read().to_arrow(plan.splits()).to_pylist() + wanted = expected if predicate is None else ([] if predicate.field == 'id' else [expected[1]]) + assert sorted(rows, key=lambda row: row['id']) == wanted + assert plan.snapshot_id == 3 + if predicate is None: + assert len(plan.splits()) == (2 if target_size == '1b' else 1) + + if engine == 'first-row' and merge_on_read == 'true': + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist([{'id': 5, 'value': 'pending'}], schema=schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + scan = table.copy({'scan.native-plan.enabled': 'true'}).new_read_builder().new_scan() + with patch.object(scan.file_scanner, 'scan', wraps=scan.file_scanner.scan) as fallback: + plan = scan.plan() + fallback.assert_called_once_with() + assert any(file.level == 0 for split in plan.splits() for file in split.files) + + [email protected]('target_size', ['1b', '1mb']) [email protected]('compacted_partition', [False, True]) [email protected]('batch_size', [1, 1024]) +def test_first_row_level_zero_merges_before_filtering(tmp_path, target_size, compacted_partition, batch_size): + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('default', True) + catalog.create_table('default.t', Schema.from_pyarrow_schema( + pa.schema([('id', pa.int64()), ('value', pa.string()), ('dt', pa.string())]), + primary_keys=['id', 'dt'], partition_keys=['dt'], options={ + 'bucket': '1', 'merge-engine': 'first-row', 'file.format': 'parquet', + 'deletion-vectors.enabled': 'true', 'deletion-vectors.merge-on-read': 'true', + 'pk-clustering-override': 'true', 'clustering.columns': 'value', + 'source.split.target-size': target_size, 'source.split.open-file-cost': '1b', + 'read.batch-size': str(batch_size), + }), False) + table = catalog.get_table('default.t') + batches = [(0, [{'id': 1, 'value': 'first', 'dt': 'pending'}, + {'id': 2, 'value': 'deleted', 'dt': 'pending'}]), + (0, [{'id': 1, 'value': 'later', 'dt': 'pending'}, + {'id': 3, 'value': 'third', 'dt': 'pending'}])] + if compacted_partition: + batches.append((1, [{'id': 4, 'value': 'compacted', 'dt': 'ready'}])) + first_file = None + for level, rows in batches: + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(rows)) + messages = writer.prepare_commit() + for message in messages: + message.new_files = [replace(file, level=level) for file in message.new_files] + if first_file is None: + first_file = message.new_files[0] + commit.commit(messages) + finally: + writer.close() + commit.close() + vector = BitmapDeletionVector() + vector.delete(1) # Remove id=2 from the first sorted L0 run. + entry = TableDeleteByRowId(table)._write_deletion_vector_index( + GenericRow(['pending'], table.partition_keys_fields), 0, {first_file.file_name: vector}) + commit = table.new_batch_write_builder().new_commit() + try: + commit.commit([CommitMessage(partition=('pending',), bucket=0, new_files=[], index_adds=[entry])]) + finally: + commit.close() + pb = table.new_read_builder().new_predicate_builder() + all_ids = [1, 3, 4] if compacted_partition else [1, 3] + for predicate, expected in [(None, all_ids), (pb.equal('value', 'later'), []), + (pb.equal('value', 'first'), [1]), (pb.equal('id', 2), [])]: + for native in (False, True): + builder = table.copy({'scan.native-plan.enabled': str(native).lower()}).new_read_builder() + builder.with_projection(['id']) + 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() + assert plan.snapshot_id == len(batches) + 1 + assert sorted(builder.new_read().to_arrow(plan.splits()).column('id').to_pylist()) == expected diff --git a/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py b/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py new file mode 100644 index 0000000000..2d837694c8 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_plan_resolved_schema_test.py @@ -0,0 +1,184 @@ +# 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. + +from contextlib import ExitStack +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import CatalogFactory, Schema +from pypaimon.common.identifier import Identifier +from pypaimon.read.native_plan import native_runtime_available +from pypaimon.schema.data_types import AtomicType +from pypaimon.schema.schema_change import SchemaChange +from pypaimon.table.file_store_table import FileStoreTable +from pypaimon.tests.native_plan_rest_test import rest_catalog # noqa: F401 + + +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_runtime_available(), reason='Rust planner required')] + + [email protected](params=[(mode, catalog) for mode in ('append', 'pk', 'de') + for catalog in ('filesystem', 'rest', 'jdbc')], ids=lambda case: '-'.join(case)) +def source(request, tmp_path): + mode, backend = request.param + if backend == 'rest': + catalog = request.getfixturevalue('rest_catalog')[0] + else: + options = {'warehouse': str(tmp_path / 'warehouse')} + if backend == 'jdbc': + options.update({'metastore': 'jdbc', 'uri': 'jdbc:sqlite:' + str(tmp_path / 'catalog.db')}) + catalog = CatalogFactory.create(options) + catalog.create_database('default', True) + schema = pa.schema([('id', pa.int64()), ('value', pa.string())]) + options = {'file.format': 'parquet'} + if mode == 'de': + options.update({'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true'}) + if mode == 'pk': + options['bucket'] = '1' + catalog.create_table('default.t', Schema.from_pyarrow_schema( + schema, options=options, primary_keys=['id'] if mode == 'pk' else []), False) + table = catalog.get_table('default.t') + _write(table, [{'id': 1, 'value': 'old'}]) + yield catalog, table + if backend == 'jdbc': + catalog.close() + + +def _write(table, rows): + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(rows)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + + +def _read(table, native, predicate=None, projection=None): + table = table.copy_without_time_travel({'scan.native-plan.enabled': str(native).lower()}) + builder = table.new_read_builder() + if predicate is not None: + builder.with_filter(predicate) + if projection is not None: + builder.with_projection(projection) + scan = builder.new_scan() + if native: + from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader + from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader + with ExitStack() as stack: + stack.enter_context(patch.object(scan.file_scanner, 'scan', + side_effect=AssertionError('native fallback'))) + stack.enter_context(patch.object(table.schema_manager, 'latest', + side_effect=AssertionError('schema reload'))) + if type(table.catalog_environment.catalog_loader) is not RESTCatalogLoader: + stack.enter_context(patch('pypaimon_rust.datafusion.PaimonCatalog', + side_effect=AssertionError('catalog reload'))) + if type(table.catalog_environment.catalog_loader) is JdbcCatalogLoader: + stack.enter_context(patch.object(JdbcCatalogLoader, 'load', + side_effect=AssertionError('JDBC connection during planning'))) + plan = scan.plan() + else: + plan = scan.plan() + rows = builder.new_read().to_arrow(plan.splits()).to_pylist() + return plan.snapshot_id, sorted(rows, key=lambda row: row['id']) + + +def _assert_parity(table, expected, snapshot_id, **kwargs): + for native in (False, True): + assert _read(table, native, **kwargs) == (snapshot_id, expected) + + +def test_stale_schema_after_column_rename(source): + catalog, stale = source + catalog.alter_table('default.t', [SchemaChange.rename_column('value', 'renamed')], False) + latest = catalog.get_table('default.t') + _write(latest, [{'id': 2, 'renamed': 'new'}]) + assert stale.table_schema.id != latest.table_schema.id + for table, name in ((stale, 'value'), (latest, 'renamed')): + _assert_parity(table, [{'id': 1, name: 'old'}, {'id': 2, name: 'new'}], 2) + predicate = table.new_read_builder().new_predicate_builder().equal(name, 'new') + _assert_parity(table, [{'id': 2}], 2, predicate=predicate, projection=['id']) + + +def test_stale_schema_does_not_confuse_readded_column(source): + catalog, stale = source + catalog.alter_table('default.t', [SchemaChange.drop_column('value')], False) + catalog.alter_table('default.t', [SchemaChange.add_column('value', AtomicType('STRING'))], False) + latest = catalog.get_table('default.t') + _write(latest, [{'id': 2, 'value': 'new'}]) + assert stale.fields[1].id != latest.fields[1].id + _assert_parity(stale, [{'id': 1, 'value': 'old'}, {'id': 2, 'value': None}], 2) + _assert_parity(latest, [{'id': 1, 'value': None}, {'id': 2, 'value': 'new'}], 2) + for table, value, id_ in ((stale, 'old', 1), (latest, 'new', 2)): + predicate = table.new_read_builder().new_predicate_builder().equal('value', value) + _assert_parity(table, [{'id': id_}], 2, predicate=predicate, projection=['id']) + + [email protected]('uri', [False, True], ids=['path', 'file-uri']) [email protected]('resolving', [False, True], ids=['local-io', 'resolving-io']) +def test_catalogless_table_uses_resolved_schema(source, uri, resolving): + from pathlib import Path + from urllib.parse import unquote, urlparse + _, table = source + path = unquote(urlparse(table.table_path).path) if table.table_path.startswith('file:') else table.table_path + location = Path(path).as_uri() if uri else path + direct = FileStoreTable.from_path(location, {'resolving-file-io.enabled': str(resolving).lower()}) + assert direct.catalog_environment.catalog_loader is None + _assert_parity(direct, [{'id': 1, 'value': 'old'}], 1) + + +def test_dotted_database_does_not_require_catalog_parsing(source): + _, table = source + resolved = FileStoreTable(table.file_io, Identifier('namespace.database', 't'), + table.table_path, table.table_schema) + _assert_parity(resolved, [{'id': 1, 'value': 'old'}], 1) + + +def test_copy_removes_search_options_and_normalizes_values(source): + catalog, table = source + catalog.alter_table('default.t', [SchemaChange.set_option('scalar-index.search-mode', 'full')], False) + table = catalog.get_table('default.t').copy({ + 'scalar-index.search-mode': None, 'read.batch-size': 1, 'metadata.stats-mode': 'none'}) + _assert_parity(table, [{'id': 1, 'value': 'old'}], 1) + + +def test_historical_schema_remains_resolved_after_selector_removal(source): + catalog, old = source + catalog.alter_table('default.t', [SchemaChange.rename_column('value', 'renamed')], False) + latest = catalog.get_table('default.t') + _write(latest, [{'id': 2, 'renamed': 'new'}]) + historical = latest.copy({'scan.snapshot-id': '1'}) + assert historical.field_names == old.field_names + resumed = historical.copy_without_time_travel({'scan.snapshot-id': None}) + _assert_parity(resumed, [{'id': 1, 'value': 'old'}, {'id': 2, 'value': 'new'}], 2) + + +def test_copy_merge_engine_changes_level_zero_visibility(tmp_path): + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('default', True) + catalog.create_table('default.t', Schema.from_pyarrow_schema( + pa.schema([('id', pa.int64()), ('value', pa.string())]), primary_keys=['id'], + options={'bucket': '1', 'file.format': 'parquet'}), False) + table = catalog.get_table('default.t') + _write(table, [{'id': 1, 'value': 'old'}]) + first_row = table.copy({'merge-engine': 'first-row'}) + _assert_parity(first_row, [], 1) + _assert_parity(first_row.copy({'merge-engine': None}), [{'id': 1, 'value': 'old'}], 1) diff --git a/paimon-python/pypaimon/tests/native_plan_rest_test.py b/paimon-python/pypaimon/tests/native_plan_rest_test.py new file mode 100644 index 0000000000..2a34ce4a29 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_plan_rest_test.py @@ -0,0 +1,151 @@ +# 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. + +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import CatalogFactory, Schema +from pypaimon.api.api_response import ConfigResponse, ErrorResponse, GetTableSnapshotResponse +from pypaimon.api.auth import BearTokenAuthProvider +from pypaimon.read.native_plan import native_runtime_available +from pypaimon.snapshot.table_snapshot import TableSnapshot +from pypaimon.tests.rest.rest_server import RESTCatalogServer + + +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_runtime_available(), reason='Rust planner required')] + + [email protected] +def rest_catalog(tmp_path): + server = RESTCatalogServer(str(tmp_path), BearTokenAuthProvider('test-token'), + ConfigResponse(defaults={'prefix': 'native-test'}), 'warehouse') + server.start() + try: + catalog = CatalogFactory.create({ + 'metastore': 'rest', 'uri': server.get_url(), 'warehouse': 'warehouse', + 'token.provider': 'bear', 'token': 'test-token', 'data-token.enabled': 'false'}) + catalog.create_database('default', True) + yield catalog, server + finally: + server.shutdown() + + [email protected] +def rest_source(rest_catalog): + from pypaimon.tests.native_plan_resolved_schema_test import _write + catalog, server = rest_catalog + catalog.create_table('default.t', Schema.from_pyarrow_schema( + pa.schema([('id', pa.int64()), ('value', pa.string())])), False) + table = catalog.get_table('default.t') + _write(table, [{'id': 1, 'value': 'old'}]) + first = table.snapshot_manager().get_latest_snapshot() + _write(table, [{'id': 2, 'value': 'new'}]) + return table, server, first + + [email protected]('response', ['first', 'empty', 'missing']) +def test_catalog_snapshot_controls_native_plan(rest_source, response): + from pypaimon.tests.native_plan_resolved_schema_test import _assert_parity + table, server, first = rest_source + if response == 'first': + reply, code = GetTableSnapshotResponse(TableSnapshot(first, 1, 0, 1, first.time_millis)), 200 + elif response == 'empty': + reply, code = GetTableSnapshotResponse(), 200 + else: + code = 404 + reply = ErrorResponse('SNAPSHOT', 't', response, code) + with patch.object(server, '_table_snapshot_handle', return_value=server._mock_response(reply, code)): + expected = [{'id': 1, 'value': 'old'}] + snapshot_id = 1 + if response in ('empty', 'missing'): + expected, snapshot_id = [], None + _assert_parity(table, expected, snapshot_id) + + [email protected]('code', [403, 404, 500, 501, 503]) +def test_catalog_snapshot_failure_never_reads_disk(rest_source, code): + from pypaimon.read.native_plan import native_plan + table, server, _ = rest_source + reply = ErrorResponse('TABLE', 't', 'snapshot unavailable', code) + with patch.object(server, '_table_snapshot_handle', return_value=server._mock_response(reply, code)): + # Test the binding directly too: a native error must not become a stale + # filesystem plan before TableScan gets a chance to fall back. + with pytest.raises(Exception, match='snapshot unavailable|does not exist|permission'): + native_plan(table) + with pytest.raises(Exception): + table.copy({'scan.native-plan.enabled': 'false'}).new_read_builder().new_scan().plan() + + [email protected]('from_tag', [False, True], ids=['empty-branch', 'tagged-branch']) +def test_rest_branch_keeps_catalog_snapshot_and_schema(rest_source, rest_catalog, from_tag): + from pypaimon.common.identifier import Identifier + from pypaimon.tests.native_plan_resolved_schema_test import _assert_parity + table, server, _ = rest_source + catalog, _ = rest_catalog + if from_tag: + catalog.create_tag(table.identifier, 'first', 1) + catalog.create_branch(table.identifier, 'dev', tag_name='first' if from_tag else None) + branch = catalog.get_table(Identifier('default', 't', branch='dev')).copy({'read.batch-size': '1'}) + with patch.object(server, '_table_snapshot_handle', wraps=server._table_snapshot_handle) as load: + _assert_parity(branch, [{'id': 1, 'value': 'old'}] if from_tag else [], 1 if from_tag else None) + assert load.call_count >= 2 + assert all(call.args[2] == 'dev' for call in load.call_args_list) + + +def test_resolved_rest_table_keeps_refreshable_file_io(rest_source, rest_catalog): + from pypaimon.api.api_response import GetTableTokenResponse + from pypaimon.read.native_plan import _resolved_schema_json + from pypaimon_rust.datafusion import PaimonCatalog + table, server, _ = rest_source + catalog, _ = rest_catalog + options = dict(catalog.context.options.to_map(), **{'data-token.enabled': 'true'}) + # Expiry is in the past, so a subsequent data access must refresh; no sleep. + expired = server._mock_response(GetTableTokenResponse(token={}, expires_at_millis=0), 200) + with patch.object(server, '_table_token_handle', return_value=expired): + rt = PaimonCatalog(options).get_table('default.t') + resolved = rt.copy_with_resolved_schema(_resolved_schema_json(table)) + assert resolved.latest_snapshot().id() == 2 + denied = server._mock_response(ErrorResponse('TABLE', 't', 'token denied', 403), 403) + with patch.object(server, '_table_token_handle', return_value=denied) as refresh: + with pytest.raises(Exception, match='token denied'): + resolved.new_read_builder().new_scan().plan() + refresh.assert_called() + + [email protected]('branch', [None, 'dev']) +def test_rest_dotted_database_and_table_keep_identity(rest_catalog, branch): + from pypaimon.common.identifier import Identifier + from pypaimon.tests.native_plan_resolved_schema_test import _assert_parity, _write + catalog, server = rest_catalog + identifier = Identifier('namespace.database', 'table.with.dots') + catalog.create_database(identifier.get_database_name(), False) + catalog.create_table(identifier, Schema.from_pyarrow_schema( + pa.schema([('id', pa.int64()), ('value', pa.string())])), False) + table = catalog.get_table(identifier) + _write(table, [{'id': 1, 'value': 'old'}]) + if branch: + catalog.create_tag(identifier, 'first', 1) + catalog.create_branch(identifier, branch, tag_name='first') + _write(table, [{'id': 2, 'value': 'main'}]) + table = catalog.get_table(Identifier('namespace.database', 'table.with.dots', branch=branch)) + with patch.object(server, '_table_snapshot_handle', wraps=server._table_snapshot_handle) as load: + _assert_parity(table, [{'id': 1, 'value': 'old'}], 1) + assert all((call.args[1].get_database_name(), call.args[1].get_table_name()) + == ('namespace.database', 'table.with.dots') for call in load.call_args_list) diff --git a/paimon-python/pypaimon/tests/native_plan_test.py b/paimon-python/pypaimon/tests/native_plan_test.py index 315e15b91e..b7892c1185 100644 --- a/paimon-python/pypaimon/tests/native_plan_test.py +++ b/paimon-python/pypaimon/tests/native_plan_test.py @@ -106,11 +106,65 @@ class NativePlanTest(unittest.TestCase): version_patcher.start() self.addCleanup(version_patcher.stop) + def test_resolved_schema_keeps_custom_io_and_rest_on_catalog_path(self): + from pypaimon.catalog.catalog_environment import CatalogEnvironment + from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader + from pypaimon.filesystem.local_file_io import LocalFileIO + from pypaimon.read.native_plan import _resolved_schema_file_io_options + + class CustomIO(LocalFileIO): + pass + + class CustomEnvironment(CatalogEnvironment): + pass + + class CustomLoader(FileSystemCatalogLoader): + pass + + class CustomJdbcLoader(JdbcCatalogLoader): + pass + + table = Mock(file_io=LocalFileIO(), catalog_environment=CatalogEnvironment.empty()) + with patch('pypaimon.read.native_plan.native_method_available', return_value=True): + self.assertEqual(_resolved_schema_file_io_options(table), {}) + table.file_io = CustomIO() + self.assertIsNone(_resolved_schema_file_io_options(table)) + table.file_io = LocalFileIO() + table.catalog_environment = CustomEnvironment() + self.assertIsNone(_resolved_schema_file_io_options(table)) + table.catalog_environment = CatalogEnvironment.empty() + for loader_type in (RESTCatalogLoader, CustomLoader, CustomJdbcLoader): + table.catalog_environment.catalog_loader = loader_type( + CatalogContext.create_from_options(Options({}))) + self.assertIsNone(_resolved_schema_file_io_options(table)) + for loader_type in (FileSystemCatalogLoader, JdbcCatalogLoader): + for attr in ('hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'): + context = CatalogContext.create_from_options(Options({})) + setattr(context, attr, object()) + table.catalog_environment.catalog_loader = loader_type(context) + self.assertIsNone(_resolved_schema_file_io_options(table)) + def test_switch_defaults_off(self): self.assertFalse(CoreOptions(Options({})).native_plan_enabled()) self.assertTrue( CoreOptions(Options({"scan.native-plan.enabled": "true"})).native_plan_enabled()) + def test_catalogless_standard_file_io_options_are_preserved(self): + from pypaimon.catalog.catalog_environment import CatalogEnvironment + from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO + from pypaimon.filesystem.resolving_file_io import ResolvingFileIO + from pypaimon.read.native_plan import _resolved_schema_file_io_options + + properties = Options({'s3.path-style-access': True, 's3.endpoint': 'http://localhost:9000'}) + # No storage connection is needed to check the resolved context transfer. + arrow = PyArrowFileIO.__new__(PyArrowFileIO) + arrow.properties = properties + for file_io in (arrow, ResolvingFileIO(properties)): + table = Mock(file_io=file_io, catalog_environment=CatalogEnvironment.empty()) + with patch('pypaimon.read.native_plan.native_method_available', return_value=True): + self.assertEqual(_resolved_schema_file_io_options(table), { + 's3.path-style-access': 'true', 's3.endpoint': 'http://localhost:9000'}) + def test_plan_uses_file_scanner_when_switch_off(self): fs = Mock() sentinel = object() @@ -281,7 +335,6 @@ class NativePlanTest(unittest.TestCase): setattr(fs, 'idx_of_this_subtask', 0))) check(lambda s, fs: (setattr(fs, 'data_evolution', True), setattr(fs, 'start_pos_of_this_subtask', 0))) - check(lambda s, fs: setattr(fs, 'chunk_shuffle', (1, 100))) check(lambda s, fs: setattr(fs, '_global_index_result', object())) check(lambda s, fs: setattr(fs, '_row_ranges', [object()])) check(lambda s, fs: setattr(fs, 'deletion_vectors_enabled', True)) @@ -293,8 +346,6 @@ class NativePlanTest(unittest.TestCase): check(lambda s, fs: setattr(s.table.schema_manager.latest.return_value, 'id', 2)) check(lambda s, fs: s.table.schema_manager.latest.__setattr__( 'side_effect', RuntimeError('metadata read failed'))) - check(lambda s, fs: s.table.options.merge_engine.__setattr__( - 'return_value', 'first-row')) check(lambda s, fs: setattr(s.table.options, 'query_auth_enabled', True)) check(lambda s, fs: s.table.current_branch.__setattr__('return_value', 'b1')) check(lambda s, fs: s.table.identifier.get_database_name.__setattr__( diff --git a/paimon-python/pypaimon/tests/reader_predicate_test.py b/paimon-python/pypaimon/tests/reader_predicate_test.py index 89e4efef6a..accce5555d 100644 --- a/paimon-python/pypaimon/tests/reader_predicate_test.py +++ b/paimon-python/pypaimon/tests/reader_predicate_test.py @@ -21,6 +21,7 @@ import tempfile import unittest import pyarrow as pa +import pytest from pypaimon import CatalogFactory from pypaimon import Schema @@ -30,6 +31,83 @@ from pypaimon.read.split import Split from pypaimon.schema.data_types import AtomicType, DataField [email protected]('mode', ['append', 'pk-raw', 'pk-merge', 'first-row']) [email protected]('batch_size', [1, 1024]) +def test_predicate_preserves_deletion_vector_positions(tmp_path, mode, batch_size): + from dataclasses import replace + from pypaimon.deletionvectors.bitmap_deletion_vector import BitmapDeletionVector + from pypaimon.table.row.generic_row import GenericRow + from pypaimon.write.commit_message import CommitMessage + from pypaimon.write.table_delete import TableDeleteByRowId + + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('default', False) + options = {'file.format': 'parquet', 'deletion-vectors.enabled': 'true', + 'deletion-vectors.merge-on-read': 'true', 'scan.native-plan.enabled': 'false', + 'read.batch-size': str(batch_size)} + if mode != 'append': + options['bucket'] = '1' + if mode == 'first-row': + options.update({'merge-engine': 'first-row', 'pk-clustering-override': 'true', + 'clustering.columns': 'id'}) + catalog.create_table('default.t', Schema.from_pyarrow_schema( + pa.schema([('id', pa.int64())]), options=options, + primary_keys=[] if mode == 'append' else ['id']), False) + table = catalog.get_table('default.t') + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pydict({'id': [0, 1, 2]})) + messages = writer.prepare_commit() + if mode == 'pk-raw': + for message in messages: + message.new_files = [replace(file, level=1) for file in message.new_files] + commit.commit(messages) + finally: + writer.close() + commit.close() + vector = BitmapDeletionVector() + vector.delete(1) + entry = TableDeleteByRowId(table)._write_deletion_vector_index( + GenericRow([], []), messages[0].bucket, {messages[0].new_files[0].file_name: vector}) + commit = builder.new_commit() + try: + commit.commit([CommitMessage(partition=(), bucket=messages[0].bucket, + new_files=[], index_adds=[entry])]) + finally: + commit.close() + for id_, expected in [(0, [0]), (1, []), (2, [2])]: + read = table.new_read_builder() + read.with_filter(read.new_predicate_builder().equal('id', id_)) + plan = read.new_scan().plan() + if mode in ('pk-merge', 'first-row'): + # A single unique L0 file can be planned raw; exercise the merge + # reader explicitly as overlapping runs and PK indexes do. + for split in plan.splits(): + split.raw_convertible = False + assert read.new_read().to_arrow(plan.splits()).column('id').to_pylist() == expected + if mode != 'append': + from copy import copy + from pypaimon.globalindex.indexed_split import IndexedSplit + from pypaimon.utils.range import Range + + # PK indexes select physical positions in one source file. Exercise + # range filtering both with and without a DV, including a deleted hit. + original = table.new_read_builder().new_scan().plan().splits()[0] + for with_dv in (False, True): + split = copy(original) + split.raw_convertible = False + if not with_dv: + split.data_deletion_files = None + for start, end in ((1, 1), (2, 2), (1, 2)): + indexed = IndexedSplit(split, [Range(start, end)]) + for id_ in (1, 2): + read = table.new_read_builder() + read.with_filter(read.new_predicate_builder().equal('id', id_)) + expected = [id_] if start <= id_ <= end and (not with_dv or id_ != 1) else [] + assert read.new_read().to_arrow([indexed]).column('id').to_pylist() == expected + + class ReaderPredicateTest(unittest.TestCase): @classmethod def setUpClass(cls): diff --git a/paimon-python/pypaimon/tests/resolving_file_io_test.py b/paimon-python/pypaimon/tests/resolving_file_io_test.py index c3dec4ebe5..f73948a0f9 100644 --- a/paimon-python/pypaimon/tests/resolving_file_io_test.py +++ b/paimon-python/pypaimon/tests/resolving_file_io_test.py @@ -65,6 +65,27 @@ class ResolvingFileIOTest(unittest.TestCase): self.assertIsInstance(fio_local, LocalFileIO) self.assertIsInstance(fio_file, LocalFileIO) + def test_pyarrow_reader_resolves_each_file_path(self): + import pyarrow as pa + import pyarrow.parquet as pq + from pathlib import Path + from pypaimon.read.reader.format_pyarrow_reader import FormatPyArrowReader + from pypaimon.schema.data_types import AtomicType, DataField + + with tempfile.TemporaryDirectory() as directory: + resolving = ResolvingFileIO(Options({})) + for value, uri in ((1, False), (2, True)): + path = Path(directory) / ('data-%s.parquet' % value) + pq.write_table(pa.table({'id': [value]}), str(path)) + reader = FormatPyArrowReader( + resolving, 'parquet', path.as_uri() if uri else str(path), + [DataField(0, 'id', AtomicType('BIGINT'))], None) + try: + self.assertEqual(reader.read_arrow_batch().to_pylist(), [{'id': value}]) + self.assertIsNone(reader.read_arrow_batch()) + finally: + reader.close() + def test_is_object_store_with_oss_warehouse(self): opts = Options({CatalogOptions.WAREHOUSE.key(): 'oss://bucket/warehouse'}) resolving = ResolvingFileIO(opts) diff --git a/paimon-python/pypaimon/tests/schema_evolution_read_test.py b/paimon-python/pypaimon/tests/schema_evolution_read_test.py index 7bbee3d84a..916058812c 100644 --- a/paimon-python/pypaimon/tests/schema_evolution_read_test.py +++ b/paimon-python/pypaimon/tests/schema_evolution_read_test.py @@ -519,11 +519,9 @@ class SchemaEvolutionReadTest(unittest.TestCase): table_read = read_builder.new_read() actual = table_read.to_arrow(splits) + # Old files supply NULL for behavior; NULL = 'g' cannot pass AND. expected = pa.Table.from_pydict({ - 'user_id': [1, 2, 4, 3, 7], - 'item_id': [1001, 1002, 1004, 1003, 1007], - 'dt': ["p1", "p1", "p1", "p2", "p2"], - 'behavior': [None, None, None, None, "g"], + 'user_id': [7], 'item_id': [1007], 'dt': ["p2"], 'behavior': ["g"], }, schema=pa_schema) self.assertEqual(expected, actual) @@ -873,3 +871,50 @@ class SchemaEvolutionReadTest(unittest.TestCase): def _scan_table(self, read_builder): splits = read_builder.new_scan().plan().splits() return splits + + [email protected]('file_format', ['parquet', 'avro']) [email protected]('evolution', ['rename', 'readd']) +def test_cross_schema_filter_runs_after_field_id_mapping(tmp_path, file_format, evolution): + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('default', True) + schema = pa.schema([('k', pa.int64()), ('v', pa.string())]) + catalog.create_table('default.t', Schema.from_pyarrow_schema( + schema, options={'file.format': file_format}), False) + + def write(table, values): + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(values)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + + write(catalog.get_table('default.t'), [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}]) + if evolution == 'rename': + catalog.alter_table('default.t', [SchemaChange.rename_column('v', 'renamed')], False) + latest_rows = [{'k': 3, 'renamed': 'c'}, {'k': 4, 'renamed': 'b'}] + else: + catalog.alter_table('default.t', [SchemaChange.drop_column('v')], False) + catalog.alter_table('default.t', [SchemaChange.add_column('v', AtomicType('STRING'))], False) + latest_rows = [{'k': 3, 'v': 'b'}, {'k': 4, 'v': None}] + table = catalog.get_table('default.t').copy({'scan.native-plan.enabled': 'false'}) + write(table, latest_rows) + # Read an unfiltered plan so manifest statistics cannot hide a reader bug. + splits = table.new_read_builder().new_scan().plan().splits() + pb = table.new_read_builder().new_predicate_builder() + cases = ([(pb.equal('renamed', 'b'), [2, 4]), (pb.equal('renamed', 'missing'), [])] + if evolution == 'rename' else [(pb.is_null('v'), [1, 2, 4]), (pb.equal('v', 'b'), [3])]) + for predicate, expected in cases: + for limit in (None, 1): + builder = table.new_read_builder().with_filter(predicate).with_projection(['k']) + if limit is not None: + builder.with_limit(limit) + actual = builder.new_read().to_arrow(splits, parallelism=1).column('k').to_pylist() + if limit is None: + assert sorted(actual) == expected + else: + assert len(actual) == min(1, len(expected)) + assert all(key in expected for key in actual) diff --git a/paimon-python/pypaimon/tests/snapshot_manager_test.py b/paimon-python/pypaimon/tests/snapshot_manager_test.py index f09f61fc3a..21dbaa7860 100644 --- a/paimon-python/pypaimon/tests/snapshot_manager_test.py +++ b/paimon-python/pypaimon/tests/snapshot_manager_test.py @@ -47,6 +47,30 @@ def _build_manager(file_io): class SnapshotManagerTest(unittest.TestCase): """Tests for SnapshotManager batch lookahead methods.""" + def test_rest_snapshot_response_can_be_empty(self): + from pypaimon.api.api_response import GetTableSnapshotResponse + from pypaimon.common.json_util import JSON + for payload in ('{}', '{"snapshot": null}'): + self.assertIsNone(JSON.from_json(payload, GetTableSnapshotResponse).get_snapshot()) + + def test_only_unsupported_rest_snapshot_loader_falls_back(self): + from pypaimon.api.rest_exception import NotImplementedException, RESTException + from pypaimon.snapshot.snapshot_loader import SnapshotLoader + from pypaimon.snapshot.snapshot_manager import SnapshotManager + catalog_loader = Mock() + catalog_loader.load.return_value.load_snapshot.side_effect = NotImplementedError('unsupported') + manager = SnapshotManager(Mock(), '/table', snapshot_loader=SnapshotLoader(catalog_loader, Mock())) + snapshot = _create_mock_snapshot(2) + manager._get_latest_snapshot_from_filesystem = Mock(return_value=snapshot) + self.assertIs(manager.get_latest_snapshot(), snapshot) + manager._get_latest_snapshot_from_filesystem.assert_called_once() + manager._get_latest_snapshot_from_filesystem.reset_mock() + for error in (RESTException('unavailable'), NotImplementedException('unavailable')): + catalog_loader.load.return_value.load_snapshot.side_effect = error + with self.assertRaisesRegex(RuntimeError, 'unavailable'): + manager.get_latest_snapshot() + manager._get_latest_snapshot_from_filesystem.assert_not_called() + def test_find_next_scannable_returns_first_matching(self): """find_next_scannable should return the first snapshot that passes should_scan.""" file_io = Mock()
