You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/09/16 05:07:14 UTC
(paimon) branch master updated: [python] Expand native planning across resolved tables and scan modes (#9873)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d1a200fee1 [python] Expand native planning across resolved tables and scan modes (#9873)
d1a200fee1 is described below
commit d1a200fee142256ece9c716012c2c25bcdb7a6e7
Author: Jingsong Lee <ji...@gmail.com>
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')]
+
+
+@pytest.fixture(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]
+
+
+@pytest.mark.parametrize('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')]
+
+
+@pytest.mark.parametrize('engine', ['deduplicate', 'first-row'])
+@pytest.mark.parametrize('merge_on_read', ['false', 'true'])
+@pytest.mark.parametrize('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)
+
+
+@pytest.mark.parametrize('target_size', ['1b', '1mb'])
+@pytest.mark.parametrize('compacted_partition', [False, True])
+@pytest.mark.parametrize('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')]
+
+
+@pytest.fixture(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'])
+
+
+@pytest.mark.parametrize('uri', [False, True], ids=['path', 'file-uri'])
+@pytest.mark.parametrize('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')]
+
+
+@pytest.fixture
+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()
+
+
+@pytest.fixture
+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
+
+
+@pytest.mark.parametrize('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)
+
+
+@pytest.mark.parametrize('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()
+
+
+@pytest.mark.parametrize('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()
+
+
+@pytest.mark.parametrize('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
+@pytest.mark.parametrize('mode', ['append', 'pk-raw', 'pk-merge', 'first-row'])
+@pytest.mark.parametrize('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
+
+
+@pytest.mark.parametrize('file_format', ['parquet', 'avro'])
+@pytest.mark.parametrize('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()