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 875cfa771d [python] Support plan by paimon-rust (#8523)
875cfa771d is described below
commit 875cfa771d2c4ee47474d254d5ae455a237487c7
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jul 20 13:01:07 2026 +0800
[python] Support plan by paimon-rust (#8523)
---
.github/workflows/paimon-python-checks.yml | 10 +-
.../pypaimon/common/options/core_options.py | 12 +
.../pypaimon/globalindex/indexed_split.py | 5 +
paimon-python/pypaimon/read/explain.py | 6 +
paimon-python/pypaimon/read/native_plan.py | 124 +++++++
paimon-python/pypaimon/read/read_builder.py | 13 +-
paimon-python/pypaimon/read/split.py | 8 +-
paimon-python/pypaimon/read/split_serializer.py | 304 +++++++++++++++++
paimon-python/pypaimon/read/table_scan.py | 145 +++++++-
paimon-python/pypaimon/table/file_store_table.py | 8 +-
.../pypaimon/tests/native_plan_integration_test.py | 230 +++++++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 379 +++++++++++++++++++++
.../pypaimon/tests/split_serializer_test.py | 170 +++++++++
paimon-python/setup.py | 2 +-
14 files changed, 1405 insertions(+), 11 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml
b/.github/workflows/paimon-python-checks.yml
index af7c7883fa..c3b35b7f4c 100755
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -36,6 +36,7 @@ env:
JDK_VERSION: 8
MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30
-Dmaven.wagon.http.retryHandler.requestSentEnabled=true
LUMINA_DATA_VERSION: 0.1.0
+ PYPAIMON_RUST_REV: 7c568274a4c5df6bd00e8d60edea21395412202b
concurrency:
@@ -142,7 +143,14 @@ jobs:
else
python -m pip install --upgrade pip
pip install torch --index-url https://download.pytorch.org/whl/cpu
- python -m pip install pyroaring readerwriterlock==1.0.9
fsspec==2024.3.1 cachetools==5.3.3 ossfs==2023.12.0 ray==2.54.0
fastavro==1.11.1 'isal>=1.8,<2' pyarrow==16.0.0 zstandard==0.24.0
polars==1.32.0 duckdb==1.3.2 numpy==1.24.3 pandas==2.0.3 pylance==0.39.0
cramjam flake8==4.0.1 pytest~=7.0 py4j==0.10.9.9 requests parameterized==0.9.0
'daft>=0.7.6' pypaimon-rust==0.2.0 'datafusion>=52'
+ python -m pip install pyroaring readerwriterlock==1.0.9
fsspec==2024.3.1 cachetools==5.3.3 ossfs==2023.12.0 ray==2.54.0
fastavro==1.11.1 'isal>=1.8,<2' pyarrow==16.0.0 zstandard==0.24.0
polars==1.32.0 duckdb==1.3.2 numpy==1.24.3 pandas==2.0.3 pylance==0.39.0
cramjam flake8==4.0.1 pytest~=7.0 py4j==0.10.9.9 requests parameterized==0.9.0
'daft>=0.7.6' 'datafusion>=52'
+ if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
+ # Exercise the split-planning API in one lane until the
compatible 0.3 wheel is published.
+ python -m pip install
"git+https://github.com/apache/paimon-rust.git@${PYPAIMON_RUST_REV}#subdirectory=bindings/python";
+ python -c "from pypaimon_rust.datafusion import PaimonCatalog;
assert hasattr(PaimonCatalog, 'get_table')"
+ else
+ python -m pip install pypaimon-rust==0.2.0
+ fi
python -m pip install 'lumina-data>=${{ env.LUMINA_DATA_VERSION
}}' -i https://pypi.org/simple/
if python -c "import sys; sys.exit(0 if sys.version_info >= (3,
11) else 1)"; then
python -m pip install vortex-data==0.70.0
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 4d2e6a7bb8..353c140ff5 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -533,6 +533,15 @@ class CoreOptions:
.with_description("Whether to enable deletion vectors.")
)
+ SCAN_NATIVE_PLAN_ENABLED: ConfigOption[bool] = (
+ ConfigOptions.key("scan.native-plan.enabled")
+ .boolean_type()
+ .default_value(False)
+ .with_description("Plan splits via the native (pypaimon_rust) planner "
+ "instead of the Python manifest scanner; the
pypaimon "
+ "reader still reads the files.")
+ )
+
CHANGELOG_PRODUCER: ConfigOption[ChangelogProducer] = (
ConfigOptions.key("changelog-producer")
.enum_type(ChangelogProducer)
@@ -1225,6 +1234,9 @@ class CoreOptions:
def deletion_vectors_enabled(self, default=None):
return self.options.get(CoreOptions.DELETION_VECTORS_ENABLED, default)
+ def native_plan_enabled(self, default=None):
+ return self.options.get(CoreOptions.SCAN_NATIVE_PLAN_ENABLED, default)
+
def changelog_producer(self, default=None):
return self.options.get(CoreOptions.CHANGELOG_PRODUCER, default)
diff --git a/paimon-python/pypaimon/globalindex/indexed_split.py
b/paimon-python/pypaimon/globalindex/indexed_split.py
index 0eaa09ebd4..18b75a0af0 100644
--- a/paimon-python/pypaimon/globalindex/indexed_split.py
+++ b/paimon-python/pypaimon/globalindex/indexed_split.py
@@ -100,6 +100,11 @@ class IndexedSplit(Split):
"""Delegate to data_split."""
return self._data_split.data_deletion_files
+ @property
+ def snapshot_id(self):
+ """Delegate to data_split."""
+ return self._data_split.snapshot_id
+
def contains_row_id(self, row_id: int) -> bool:
"""Check if the given row ID is in the row ranges."""
for r in self._row_ranges:
diff --git a/paimon-python/pypaimon/read/explain.py
b/paimon-python/pypaimon/read/explain.py
index 29deae951f..5cc1cb1bd8 100644
--- a/paimon-python/pypaimon/read/explain.py
+++ b/paimon-python/pypaimon/read/explain.py
@@ -130,6 +130,10 @@ class ExplainResult:
# Verbose-only
splits: Optional[List[ExplainSplitInfo]] = None
+ # Native (pypaimon_rust) plan; pruning funnel not tracked. Kept last to
+ # preserve positional-arg compatibility.
+ native_planned: bool = False
+
def __str__(self) -> str:
return render_explain(self)
@@ -163,6 +167,8 @@ def render_explain(result: ExplainResult) -> str:
_line(out, "Limit", str(result.limit) if result.limit is not None else
"<none>")
out.write("\n")
+ if result.native_planned:
+ _line(out, "Planner", "native (pypaimon_rust); pruning not tracked")
_line(out, "Partition pruning",
result.partition_pruning.format() if result.partition_pruning else
"n/a")
_line(out, "Bucket pruning",
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
new file mode 100644
index 0000000000..91f9827940
--- /dev/null
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -0,0 +1,124 @@
+# 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.
+
+"""Plan splits with pypaimon_rust, decoded for the normal pypaimon reader.
+
+Optional, lazily-imported dependency; enabled by ``scan.native-plan.enabled``.
+The predicate is applied pypaimon-side (partition pruning + row/limit filter),
+so results match the normal path.
+"""
+
+from typing import List, Optional
+
+from pypaimon.common.options.config import CatalogOptions
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options_utils import OptionsUtils
+from pypaimon.read.split import Split
+from pypaimon.read.split_serializer import deserialize_split_v1
+
+
+def native_runtime_available() -> bool:
+ """Whether an installed pypaimon-rust exposes the full split-planning API.
+
+ Both entry points used by :func:`native_plan` are probed: ``PaimonCatalog.
+ get_table`` (0.3.0) and ``Split.serialize`` (the split wire format). An
+ intermediate build missing either must fall back, not fail mid-plan.
+ """
+ try:
+ from pypaimon_rust.datafusion import PaimonCatalog, Split
+ except ImportError:
+ return False
+ return hasattr(PaimonCatalog, 'get_table') and hasattr(Split, 'serialize')
+
+
+def _partition_fields(table):
+ """Ordered partition DataFields, used to decode the split partition
bytes."""
+ schema = table.table_schema
+ by_name = {f.name: f for f in schema.fields}
+ return [by_name[name] for name in schema.partition_keys]
+
+
+def _catalog_metastore(loader) -> Optional[str]:
+ """Return the Rust catalog kind for an exact built-in loader."""
+ from pypaimon.catalog.filesystem_catalog_loader import
FileSystemCatalogLoader
+ from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+
+ # Subclasses may override load() with routing or option semantics which
+ # cannot be reproduced from context().options alone.
+ if type(loader) is FileSystemCatalogLoader:
+ return 'filesystem'
+ if type(loader) is RESTCatalogLoader:
+ return 'rest'
+ return None
+
+
+def _option_value_to_string(value) -> str:
+ """Stringify an option value for the Rust catalog.
+
+ Python bools stringify to ``'True'``/``'False'``; Rust parses booleans
+ case-sensitively, so emit lowercase ``'true'``/``'false'`` instead.
+ """
+ if isinstance(value, bool):
+ return 'true' if value else 'false'
+ return OptionsUtils.convert_to_string(value)
+
+
+def _catalog_options(table) -> dict:
+ """Catalog options that built this table, to reconstruct the Rust
catalog."""
+ 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)")
+ 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
+ return normalized
+
+
+def _read_options(table) -> dict:
+ """Effective split-shaping options, including FileStoreTable.copy
overrides."""
+ return {
+ CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): str(
+ table.options.source_split_target_size()),
+ CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): str(
+ table.options.source_split_open_file_cost()),
+ }
+
+
+def native_plan(table) -> List[Split]:
+ """Plan with pypaimon_rust and return the decoded pypaimon splits.
+
+ Predicate/limit are not pushed to the native planner (pushdown is a
+ follow-up); pypaimon applies them at read time.
+ """
+ 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())
+ rust_splits =
rt.new_read_builder(_read_options(table)).new_scan().plan().splits()
+ pfields = _partition_fields(table)
+ # Trimmed primary keys decode per-file min/max keys (PK merge-on-read).
+ kfields = table.trimmed_primary_keys_fields
+ return [deserialize_split_v1(s.serialize(), pfields, kfields) for s in
rust_splits]
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index b7eebc2ffd..a531d5d2c8 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -216,9 +216,15 @@ def _build_explain_result(table, scan: TableScan, plan,
stats: ScanStats,
table_schema = table.table_schema
bucket_mode_str = _safe_bucket_mode(table)
- partition_pruning = _partition_pruning(stats, scan)
- bucket_pruning = _bucket_pruning(stats, scan)
- file_skipping = _file_skipping(stats, scan)
+ # stats is None when planned natively (pypaimon_rust): no manifest pruning
+ # funnel is tracked, so the split-level signals below are all we can
report.
+ native_planned = stats is None
+ if native_planned:
+ partition_pruning = bucket_pruning = file_skipping = None
+ else:
+ partition_pruning = _partition_pruning(stats, scan)
+ bucket_pruning = _bucket_pruning(stats, scan)
+ file_skipping = _file_skipping(stats, scan)
files_per_split = [len(getattr(s, 'files', []) or []) for s in splits]
sizes = [int(getattr(s, 'file_size', 0) or 0) for s in splits]
@@ -314,6 +320,7 @@ def _build_explain_result(table, scan: TableScan, plan,
stats: ScanStats,
split_size_p50=sz_p50,
split_size_p95=sz_p95,
has_auth=plan_has_auth,
+ native_planned=native_planned,
splits=split_infos if verbose else None,
)
diff --git a/paimon-python/pypaimon/read/split.py
b/paimon-python/pypaimon/read/split.py
index c318a15b2c..bdfc537af1 100644
--- a/paimon-python/pypaimon/read/split.py
+++ b/paimon-python/pypaimon/read/split.py
@@ -78,13 +78,16 @@ class DataSplit(Split):
partition: GenericRow,
bucket: int,
raw_convertible: bool = False,
- data_deletion_files: Optional[List[DeletionFile]] = None
+ data_deletion_files: Optional[List[DeletionFile]] = None,
+ snapshot_id: Optional[int] = None
):
self._files = files
self._partition = partition
self._bucket = bucket
self.raw_convertible = raw_convertible
self.data_deletion_files = data_deletion_files
+ # Scanned snapshot; None unless populated (e.g. by the native planner).
+ self.snapshot_id = snapshot_id
@property
def files(self) -> List[DataFileMeta]:
@@ -121,7 +124,8 @@ class DataSplit(Split):
partition=self._partition,
bucket=self._bucket,
raw_convertible=self.raw_convertible,
- data_deletion_files=filtered_data_deletion_files
+ data_deletion_files=filtered_data_deletion_files,
+ snapshot_id=self.snapshot_id
)
@property
diff --git a/paimon-python/pypaimon/read/split_serializer.py
b/paimon-python/pypaimon/read/split_serializer.py
new file mode 100644
index 0000000000..4b50c86c28
--- /dev/null
+++ b/paimon-python/pypaimon/read/split_serializer.py
@@ -0,0 +1,304 @@
+# 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.
+
+"""Deserialize the cross-language ``SplitSerializer`` v1 binary into a pypaimon
+:class:`DataSplit`.
+
+Mirror of the Java ``DataSplit#serialize`` (VERSION 8) frame wrapped in the
+``SplitSerializer`` v1 header, as produced by ``pypaimon_rust``'s
+``Split.serialize()``. Extracts the fields the reader needs, plus per-file
+min/max keys for PK merge-on-read; key/value stats (planning-only) stay empty.
+"""
+
+import struct
+from typing import List, Optional
+
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.split import DataSplit, Split
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.binary_row import BinaryRow
+from pypaimon.table.row.generic_row import GenericRow, GenericRowDeserializer
+from pypaimon.table.source.deletion_file import DeletionFile
+from pypaimon.utils.range import Range
+
+# Frame magics/versions, mirroring the Rust/Java constants.
+_SPLIT_SER_MAGIC = 0x53504C49545F5631 # "SPLIT_V1"
+_SPLIT_SER_VERSION = 1
+_TYPE_DATA_SPLIT = 1
+_TYPE_INDEXED_SPLIT = 3
+_INDEXED_SPLIT_MAGIC = -938472394838495695
+_INDEXED_SPLIT_VERSION = 1
+_SPLIT_MAGIC = -2394839472490812314
+_SPLIT_VERSION = 8
+
+_DFM_ARITY = 20
+
+
+def _f(idx, name, dtype):
+ return DataField(idx, name, dtype)
+
+
+# DataFileMeta 20-field layout (order/types mirror
DataFileMetaSerializer#toRow).
+# Fields 3/4 (min/max key) are decoded for PK tables; 5/6 (stats) stay unread.
+_DFM_FIELDS: List[DataField] = [
+ _f(0, '_FILE_NAME', AtomicType('STRING')),
+ _f(1, '_FILE_SIZE', AtomicType('BIGINT')),
+ _f(2, '_ROW_COUNT', AtomicType('BIGINT')),
+ _f(3, '_MIN_KEY', AtomicType('BYTES')),
+ _f(4, '_MAX_KEY', AtomicType('BYTES')),
+ _f(5, '_KEY_STATS', AtomicType('BYTES')),
+ _f(6, '_VALUE_STATS', AtomicType('BYTES')),
+ _f(7, '_MIN_SEQUENCE_NUMBER', AtomicType('BIGINT')),
+ _f(8, '_MAX_SEQUENCE_NUMBER', AtomicType('BIGINT')),
+ _f(9, '_SCHEMA_ID', AtomicType('BIGINT')),
+ _f(10, '_LEVEL', AtomicType('INT')),
+ _f(11, '_EXTRA_FILES', AtomicType('BYTES')), # BinaryArray<string>,
decoded below
+ _f(12, '_CREATION_TIME', AtomicType('BIGINT')), # compact millis; read
raw, wrap Timestamp
+ _f(13, '_DELETE_ROW_COUNT', AtomicType('BIGINT')),
+ _f(14, '_EMBEDDED_FILE_INDEX', AtomicType('BYTES')),
+ _f(15, '_FILE_SOURCE', AtomicType('TINYINT')),
+ _f(16, '_VALUE_STATS_COLS', AtomicType('BYTES')),
+ _f(17, '_EXTERNAL_PATH', AtomicType('STRING')),
+ _f(18, '_FIRST_ROW_ID', AtomicType('BIGINT')),
+ _f(19, '_WRITE_COLS', AtomicType('BYTES')),
+]
+
+# Arity is fixed by DataSplit VERSION 8; keep the field list and arity in
lockstep.
+assert len(_DFM_FIELDS) == _DFM_ARITY
+
+
+def _decode_str_array(b: Optional[bytes]) -> Optional[List[str]]:
+ """Decode a Paimon ``BinaryArray<string>`` (non-null elements)."""
+ if not b:
+ return [] if b == b'' else None
+ n = struct.unpack_from('<i', b, 0)[0]
+ header = 4 + ((n + 31) // 32) * 4 # count + null bitset
+ out = []
+ for k in range(n):
+ eo = header + k * 8
+ if b[eo + 7] & 0x80: # inline: value in first 7 bytes, len in low 7
bits of byte 7
+ length = b[eo + 7] & 0x7F
+ out.append(b[eo:eo + length].decode('utf-8'))
+ else: # pointer: (var_off << 32) | len
+ slot = struct.unpack_from('<q', b, eo)[0]
+ var_off = (slot >> 32) & 0xFFFFFFFF
+ length = slot & 0xFFFFFFFF
+ out.append(b[var_off:var_off + length].decode('utf-8'))
+ return out
+
+
+def _decode_modified_utf8(raw: bytes) -> str:
+ """Decode Java writeUTF modified UTF-8 (NUL as C0 80, 3-byte for >= U+0800;
+ supplementary chars as a UTF-16 surrogate pair, recombined below)."""
+ units = []
+ i, n = 0, len(raw)
+ while i < n:
+ b = raw[i]
+ if b < 0x80:
+ units.append(b)
+ i += 1
+ elif b & 0xE0 == 0xC0:
+ units.append(((b & 0x1F) << 6) | (raw[i + 1] & 0x3F))
+ i += 2
+ else: # 0xE0: 3-byte
+ units.append(((b & 0x0F) << 12) | ((raw[i + 1] & 0x3F) << 6) |
(raw[i + 2] & 0x3F))
+ i += 3
+ out = []
+ j, m = 0, len(units)
+ while j < m:
+ c = units[j]
+ if 0xD800 <= c <= 0xDBFF and j + 1 < m and 0xDC00 <= units[j + 1] <=
0xDFFF:
+ out.append(0x10000 + ((c - 0xD800) << 10) + (units[j + 1] -
0xDC00))
+ j += 2
+ else:
+ out.append(c)
+ j += 1
+ return ''.join(chr(c) for c in out)
+
+
+class _Reader:
+ def __init__(self, data: bytes):
+ self.d = data
+ self.p = 0
+
+ def i32(self) -> int:
+ v = struct.unpack_from('>i', self.d, self.p)[0]
+ self.p += 4
+ return v
+
+ def i64(self) -> int:
+ v = struct.unpack_from('>q', self.d, self.p)[0]
+ self.p += 8
+ return v
+
+ def f32(self) -> float:
+ v = struct.unpack_from('>f', self.d, self.p)[0]
+ self.p += 4
+ return v
+
+ def u8(self) -> int:
+ v = self.d[self.p]
+ self.p += 1
+ return v
+
+ def take(self, n: int) -> bytes:
+ v = self.d[self.p:self.p + n]
+ self.p += n
+ return v
+
+ def java_utf(self) -> str:
+ n = struct.unpack_from('>H', self.d, self.p)[0]
+ self.p += 2
+ return _decode_modified_utf8(self.take(n))
+
+
+def deserialize_split_v1(data: bytes, partition_fields: List[DataField],
+ key_fields: Optional[List[DataField]] = None) ->
Split:
+ """Rebuild a pypaimon ``DataSplit`` (or ``IndexedSplit``) from
``Split.serialize()`` bytes.
+
+ ``key_fields`` (trimmed primary keys) decode per-file min/max keys for PK
+ merge-on-read; None for append tables.
+ """
+ r = _Reader(data)
+ magic = r.i64()
+ if magic != _SPLIT_SER_MAGIC:
+ raise ValueError("bad SplitSerializer magic %d" % magic)
+ version = r.i32()
+ if version != _SPLIT_SER_VERSION:
+ raise ValueError(
+ "unsupported SplitSerializer version %d (expected %d)" % (version,
_SPLIT_SER_VERSION))
+ type_id = r.i32()
+ if type_id == _TYPE_DATA_SPLIT:
+ return _read_datasplit_body(r, partition_fields, key_fields)
+ if type_id == _TYPE_INDEXED_SPLIT:
+ imagic = r.i64()
+ if imagic != _INDEXED_SPLIT_MAGIC:
+ raise ValueError("bad IndexedSplit magic %d" % imagic)
+ iversion = r.i32()
+ if iversion != _INDEXED_SPLIT_VERSION:
+ raise ValueError(
+ "unsupported IndexedSplit version %d (expected %d)" %
(iversion, _INDEXED_SPLIT_VERSION))
+ data_split = _read_datasplit_body(r, partition_fields, key_fields)
+ # row_ranges select which global row ids to read -- must be preserved,
+ # else the reader scans the whole file instead of the ANN/row-id
result.
+ row_ranges = [Range(r.i64(), r.i64()) for _ in range(r.i32())]
+ return IndexedSplit(data_split, row_ranges, _read_scores(r))
+ raise ValueError("unsupported split type id %d" % type_id)
+
+
+def _read_scores(r: '_Reader') -> Optional[List[float]]:
+ if r.u8() == 0:
+ return None
+ return [r.f32() for _ in range(r.i32())]
+
+
+def _read_datasplit_body(r: _Reader, partition_fields: List[DataField],
+ key_fields: Optional[List[DataField]] = None) ->
DataSplit:
+ if r.i64() != _SPLIT_MAGIC:
+ raise ValueError("bad DataSplit magic")
+ version = r.i32()
+ if version != _SPLIT_VERSION:
+ raise ValueError(
+ "unsupported DataSplit version %d (expected %d)" % (version,
_SPLIT_VERSION))
+ snapshot_id = r.i64() # scanned snapshot; row-id conflict detection
needs it
+ partition = GenericRowDeserializer.from_bytes(r.take(r.i32()),
partition_fields)
+ bucket = r.i32()
+ bucket_path = r.java_utf()
+ if r.u8() == 1:
+ r.i32() # total_buckets
+ if r.i32() != 0: # deprecated beforeFiles; Java
rejects non-empty
+ raise ValueError("cannot deserialize a split with before files")
+ if r.u8() != 0: # beforeDeletionFiles must be null
+ raise ValueError("cannot deserialize a split with before deletion
files")
+ file_count = r.i32()
+ files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, key_fields)
+ for _ in range(file_count)]
+ data_deletion_files = _read_deletion_list(r)
+ r.u8() # isStreaming
+ raw_convertible = r.u8() != 0
+ return DataSplit(
+ files=files,
+ partition=partition,
+ bucket=bucket,
+ raw_convertible=raw_convertible,
+ data_deletion_files=data_deletion_files,
+ snapshot_id=snapshot_id,
+ )
+
+
+def _decode_key(b: Optional[bytes], key_fields: Optional[List[DataField]]) ->
GenericRow:
+ """Decode a min/max key (serialized BinaryRow); empty for append tables."""
+ if not key_fields or not b:
+ return GenericRow([], [])
+ return GenericRowDeserializer.from_bytes(b, key_fields)
+
+
+def _datafilemeta_from_row(row_bytes: bytes, bucket_path: str,
+ key_fields: Optional[List[DataField]] = None) ->
DataFileMeta:
+ row = BinaryRow(struct.pack('>i', _DFM_ARITY) + row_bytes, _DFM_FIELDS)
+ g = row.get_field
+ file_name = g(0)
+ external_path = g(17)
+ ct = g(12)
+ ct = Timestamp(int(ct)) if ct is not None else None
+ # min/max keys drive PK merge-on-read; stats unused, left empty.
+ meta = DataFileMeta(
+ file_name=file_name,
+ file_size=g(1),
+ row_count=g(2),
+ min_key=_decode_key(g(3), key_fields),
+ max_key=_decode_key(g(4), key_fields),
+ key_stats=SimpleStats.empty_stats(),
+ value_stats=SimpleStats.empty_stats(),
+ min_sequence_number=g(7),
+ max_sequence_number=g(8),
+ schema_id=g(9),
+ level=g(10),
+ extra_files=_decode_str_array(g(11)) or [],
+ creation_time=ct,
+ delete_row_count=g(13),
+ embedded_index=g(14),
+ file_source=g(15),
+ value_stats_cols=_decode_str_array(g(16)),
+ external_path=external_path,
+ first_row_id=g(18),
+ write_cols=_decode_str_array(g(19)),
+ )
+ meta.file_path = external_path if external_path else "%s/%s" %
(bucket_path.rstrip('/'), file_name)
+ return meta
+
+
+def _read_deletion_list(r: _Reader) -> Optional[List[Optional[DeletionFile]]]:
+ if r.u8() == 0:
+ return None
+ result: List[Optional[DeletionFile]] = []
+ for _ in range(r.i32()):
+ if r.u8() == 0:
+ result.append(None)
+ continue
+ path = r.java_utf()
+ offset, length, cardinality = r.i64(), r.i64(), r.i64()
+ result.append(DeletionFile(
+ dv_index_path=path,
+ offset=offset,
+ length=length,
+ cardinality=None if cardinality == -1 else cardinality,
+ ))
+ return result
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 8c89fc3b76..313046aecb 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -16,9 +16,11 @@
# under the License.
import json as _json
+import logging
from typing import Optional, Tuple
from pypaimon.catalog.catalog_exception import TableNoPermissionException
+from pypaimon.common.identifier import UNKNOWN_DATABASE
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
@@ -28,6 +30,15 @@ from pypaimon.read.query_auth_split import
resolve_auth_result, wrap_plan_with_a
from pypaimon.read.scan_stats import ScanStats
from pypaimon.read.scanner.file_scanner import FileScanner
+logger = logging.getLogger(__name__)
+
+# Options native forwards to Rust; any other copy() override is invisible to
Rust.
+_NATIVE_FORWARDED_OPTIONS = frozenset({
+ CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key(),
+ CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(),
+ CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(),
+})
+
class TableScan:
"""Implementation of TableScan for native Python reading."""
@@ -52,11 +63,134 @@ class TableScan:
def plan(self) -> Plan:
auth_result = self.__auth_query()
+ # Native planning covers only a plain full-snapshot scan and bypasses
the
+ # auth-aware file scanner; fall back to the normal path otherwise.
+ if (auth_result is None and self.table.options.native_plan_enabled()
+ and self._native_plan_supported()):
+ native = self._try_native_plan()
+ if native is not None:
+ return native
if auth_result is not None:
prune_scanner_by_auth(self.table, self.file_scanner, auth_result)
plan = self.file_scanner.scan()
return wrap_plan_with_auth(auth_result, plan)
+ def _native_plan_supported(self) -> bool:
+ # Any probe failure (e.g. a remote schema/metadata read) must fall
back, not fail the scan.
+ try:
+ return self._native_plan_supported_impl()
+ except Exception as e:
+ logger.warning("Native-plan capability probe failed, falling back:
%s", e)
+ return False
+
+ def _native_plan_supported_impl(self) -> bool:
+ """Fall back to the Python scanner for scans native can't carry:
+ shard/slice, chunk-shuffle, global-index, first-row merge-engine (Rust
+ drops L0), deletion vectors (Python drops L0), data evolution
+ (dedicated split generator), postpone bucket (drops synthetic buckets),
+ a primary-key table whose trimmed PK is empty (PK equals the partition
+ key; native may mark splits raw-convertible and skip merge), dynamic
+ bucket / cross-partition PK tables (unconfirmed Rust parity), any
+ partitioned table (Rust bucket_path vs the writer's str(value) can
+ diverge), a stale schema (Rust reloads the latest), copy() overrides
+ Rust does not see (e.g. a removed scan.snapshot-id), a row limit
+ (native has no plan-time limit pushdown), query auth, non-main branch,
+ time-travel, scan.version, incremental, 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_runtime_available
+ if not native_runtime_available():
+ return False
+ # Native has no plan-time limit pushdown; Python trims splits before
read.
+ if self.limit is not None:
+ return False
+ fs = self.file_scanner
+ if (getattr(fs, 'idx_of_this_subtask', None) is not None
+ or getattr(fs, 'start_pos_of_this_subtask', None) is not None
+ or getattr(fs, 'chunk_shuffle', None) is not None
+ or getattr(fs, '_global_index_result', None) is not None
+ or getattr(fs, 'deletion_vectors_enabled', False)
+ or getattr(fs, 'data_evolution', False)
+ or getattr(fs, 'only_read_real_buckets', False)):
+ 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
+ if self.table.options.query_auth_enabled \
+ or self.table.options.merge_engine() == 'first-row' \
+ or self.table.current_branch() != 'main':
+ return False
+ # Empty trimmed PK (PK == partition key): native skips merge ->
duplicate/stale rows.
+ if getattr(self.table, 'is_primary_key_table', False) \
+ and not self.table.trimmed_primary_keys:
+ return False
+ # Dynamic-bucket / cross-partition PK: Rust parity unconfirmed -> fall
back.
+ from pypaimon.table.bucket_mode import BucketMode
+ if self.table.bucket_mode() in (BucketMode.HASH_DYNAMIC,
BucketMode.CROSS_PARTITION):
+ return False
+ # Rust bucket_path vs the writer's unescaped str(value) can diverge ->
fall back.
+ if self.table.partition_keys:
+ return False
+ # Rust reloads the latest schema; fall back if this table's schema is
stale.
+ latest_schema = self.table.schema_manager.latest()
+ if latest_schema is not None and latest_schema.id !=
self.table.table_schema.id:
+ return False
+ # copy() overrides Rust can't see (e.g. removed scan.snapshot-id) ->
fall back.
+ overrides = set(getattr(self.table, '_applied_dynamic_options', {}) or
{})
+ if overrides - _NATIVE_FORWARDED_OPTIONS:
+ return False
+ from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+ options = self.table.options.options
+ if any(options.contains_key(k) for k in SCAN_KEYS) \
+ or options.contains_key('scan.version'):
+ return False
+ return not options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
+
+ def _try_native_plan(self) -> Optional[Plan]:
+ """Plan via pypaimon_rust, then drop partitions the predicate rejects.
+
+ The predicate is not pushed to the native planner, so this may read
more
+ files; the reader's row filter/limit still apply, so results match.
Return
+ None when Rust finds no splits so the caller can use the matching
Python
+ fallback (with scan stats when requested).
+ """
+ from pypaimon.read.native_plan import native_plan
+
+ try:
+ splits = native_plan(self.table)
+ if not splits:
+ return None
+ snapshot_id = splits[0].snapshot_id
+ partition_predicate = self.file_scanner.partition_key_predicate
+ if partition_predicate is not None:
+ splits = [s for s in splits
+ if getattr(s, 'partition', None) is None
+ or partition_predicate.test(s.partition)]
+ return Plan(splits, snapshot_id=snapshot_id)
+ except Exception as e:
+ # Any native construction/planning/pruning failure -> fall back.
+ logger.warning(
+ "Native plan failed, falling back to the Python scanner: %s",
e)
+ return None
+
def plan_for_write(self) -> Plan:
if self.__auth_query() is not None:
raise TableNoPermissionException(self.table.identifier)
@@ -65,13 +199,20 @@ class TableScan:
def __auth_query(self):
return resolve_auth_result(self._query_auth_fn, self._read_type)
- def scan_with_stats(self) -> Tuple[Plan, ScanStats]:
+ def scan_with_stats(self) -> Tuple[Plan, Optional[ScanStats]]:
"""Run :meth:`plan` while recording manifest / pruning counters.
Only used by :meth:`ReadBuilder.explain`; the regular read path
- keeps going through :meth:`plan`.
+ keeps going through :meth:`plan`. Native planning is not tracked, so
+ stats is None on the native path -- explain reflects the real plan and
+ marks the pruning funnel as untracked.
"""
auth_result = self.__auth_query()
+ if (auth_result is None and self.table.options.native_plan_enabled()
+ and self._native_plan_supported()):
+ native = self._try_native_plan()
+ if native is not None:
+ return native, None
if auth_result is not None:
prune_scanner_by_auth(self.table, self.file_scanner, auth_result)
plan, stats = self.file_scanner.scan_with_stats()
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 56e2b9518c..8edbc14021 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -520,8 +520,12 @@ class FileStoreTable(Table):
)
catalog_env = self.catalog_environment.copy(new_identifier)
- return FileStoreTable(self.file_io, new_identifier, self.table_path,
new_table_schema,
- catalog_env)
+ new_table = FileStoreTable(self.file_io, new_identifier,
self.table_path,
+ new_table_schema, catalog_env)
+ # Cumulative copy() overrides (removals kept as None) vs the on-disk
schema.
+ new_table._applied_dynamic_options = {
+ **getattr(self, '_applied_dynamic_options', {}), **options}
+ return new_table
def _try_time_travel(self, options: Options) -> Optional[TableSchema]:
"""
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
new file mode 100644
index 0000000000..a78e1b55a0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -0,0 +1,230 @@
+# 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 tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+
+
+def _has_native_planner():
+ try:
+ from pypaimon_rust.datafusion import PaimonCatalog, Split
+ except Exception:
+ return False
+ return hasattr(PaimonCatalog, 'get_table') and hasattr(Split, 'serialize')
+
+
[email protected](_has_native_planner(),
+ "pypaimon_rust with split-planning API not installed")
+class NativePlanIntegrationTest(unittest.TestCase):
+ """Live round-trip guarding the cross-language SplitSerializer against
drift:
+ plan via pypaimon_rust, decode, and require the same rows as the normal
plan.
+ The golden unit tests only prove self-consistency; this proves byte-compat
+ with the real producer."""
+
+ def setUp(self):
+ self.cat = CatalogFactory.create({'warehouse':
tempfile.mkdtemp(prefix='np_it_')})
+ self.cat.create_database('default', True)
+ self.schema = pa.schema([('k', pa.int64()), ('v', pa.string())])
+
+ def _write(self, name, rows):
+ t = self.cat.get_table('default.%s' % name)
+ wb = t.new_batch_write_builder()
+ w, c = wb.new_write(), wb.new_commit()
+ w.write_arrow(pa.Table.from_pylist(rows, schema=self.schema))
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ def _plan_and_read(self, name, native):
+ t = self.cat.get_table('default.%s' % name)
+ if native:
+ t = t.copy({'scan.native-plan.enabled': 'true'})
+ rb = t.new_read_builder()
+ plan = rb.new_scan().plan()
+ rows = rb.new_read().to_arrow(plan.splits()).to_pylist()
+ return plan.snapshot_id, sorted(rows, key=lambda r: r['k'])
+
+ def _assert_matches(self, name, expect_native=True):
+ sid_n, rows_n = self._plan_and_read(name, native=False)
+ sid_r, rows_r = self._plan_and_read(name, native=True)
+ self.assertEqual(rows_r, rows_n)
+ self.assertEqual(sid_r, sid_n) # snapshot id preserved through
native plan
+ self.assertIsNotNone(sid_r)
+ # Guard against a false green where native silently fell back to
Python: assert the
+ # native planner was (or was not) actually used, as expected for this
table.
+ native_table = self.cat.get_table('default.%s' % name).copy(
+ {'scan.native-plan.enabled': 'true'})
+ self.assertEqual(
+ native_table.new_read_builder().explain().native_planned,
expect_native)
+
+ def test_primary_key_matches_normal_plan(self):
+ self.cat.create_table('default.pk_t', Schema.from_pyarrow_schema(
+ self.schema, primary_keys=['k'], options={'bucket': '1'}), False)
+ self._write('pk_t', [{'k': 1, 'v': 'a1'}, {'k': 2, 'v': 'b1'}])
+ self._write('pk_t', [{'k': 2, 'v': 'b2'}, {'k': 3, 'v': 'c1'}]) # k=2
updated
+ self._assert_matches('pk_t')
+
+ def test_pk_equal_to_partition_key_falls_back(self):
+ # Empty trimmed PK: native would skip merge and return duplicates ->
must fall back.
+ self.cat.create_table('default.pkpart_t', Schema.from_pyarrow_schema(
+ self.schema, partition_keys=['k'], primary_keys=['k'],
options={'bucket': '1'}), False)
+ self._write('pkpart_t', [{'k': 1, 'v': 'a1'}, {'k': 2, 'v': 'b1'}])
+ self._write('pkpart_t', [{'k': 2, 'v': 'b2'}])
+ native_table = self.cat.get_table('default.pkpart_t').copy(
+ {'scan.native-plan.enabled': 'true'})
+
self.assertFalse(native_table.new_read_builder().explain().native_planned)
+ with self.assertRaises(ValueError):
+ 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.
+ 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)
+
+ def test_first_row_merge_engine_falls_back(self):
+ self.cat.create_table('default.fr_t', Schema.from_pyarrow_schema(
+ self.schema, primary_keys=['k'],
+ options={'bucket': '1', 'merge-engine': 'first-row'}), False)
+ self._write('fr_t', [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}])
+ self._write('fr_t', [{'k': 1, 'v': 'X'}, {'k': 3, 'v': 'c'}]) # k=1
stays 'a'
+ self._assert_matches('fr_t', expect_native=False)
+
+ def test_append_matches_normal_plan(self):
+ self.cat.create_table(
+ 'default.ap_t', Schema.from_pyarrow_schema(self.schema), False)
+ self._write('ap_t', [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}])
+ self._write('ap_t', [{'k': 3, 'v': 'c'}])
+ self._assert_matches('ap_t')
+
+ def test_dynamic_split_target_size_matches_normal_plan(self):
+ self.cat.create_table(
+ 'default.split_t', Schema.from_pyarrow_schema(self.schema), False)
+ self._write('split_t', [{'k': 1, 'v': 'a'}])
+ self._write('split_t', [{'k': 2, 'v': 'b'}])
+ options = {'source.split.target-size': '1b'}
+ normal_table = self.cat.get_table('default.split_t').copy(options)
+ native_table = normal_table.copy({'scan.native-plan.enabled': 'true'})
+
+ normal = normal_table.new_read_builder().new_scan().plan()
+ native = native_table.new_read_builder().explain()
+
+ self.assertTrue(native.native_planned)
+ self.assertEqual(native.split_count, len(normal.splits()))
+ self.assertGreater(native.split_count, 1)
+
+ def test_dynamic_split_open_file_cost_matches_normal_plan(self):
+ stored_options = {
+ 'source.split.target-size': '128mb',
+ 'source.split.open-file-cost': '1b',
+ }
+ self.cat.create_table('default.open_cost_t',
Schema.from_pyarrow_schema(
+ self.schema, options=stored_options), False)
+ self._write('open_cost_t', [{'k': 1, 'v': 'a'}])
+ self._write('open_cost_t', [{'k': 2, 'v': 'b'}])
+ self._write('open_cost_t', [{'k': 3, 'v': 'c'}])
+ base_table = self.cat.get_table('default.open_cost_t')
+ normal_table = base_table.copy({'source.split.open-file-cost': '64mb'})
+ native_table = normal_table.copy({'scan.native-plan.enabled': 'true'})
+
+ baseline = base_table.new_read_builder().new_scan().plan()
+ normal = normal_table.new_read_builder().new_scan().plan()
+ native = native_table.new_read_builder().explain()
+
+ self.assertEqual(len(baseline.splits()), 1)
+ self.assertGreater(len(normal.splits()), len(baseline.splits()))
+ self.assertTrue(native.native_planned)
+ self.assertEqual(native.split_count, len(normal.splits()))
+
+ def test_dynamic_split_option_reset_matches_normal_plan(self):
+ stored_options = {
+ 'source.split.target-size': '1b',
+ 'source.split.open-file-cost': '1b',
+ }
+ self.cat.create_table('default.split_reset_t',
Schema.from_pyarrow_schema(
+ self.schema, options=stored_options), False)
+ self._write('split_reset_t', [{'k': 1, 'v': 'a'}])
+ self._write('split_reset_t', [{'k': 2, 'v': 'b'}])
+ reset_options = {
+ 'source.split.target-size': None,
+ 'source.split.open-file-cost': None,
+ }
+ normal_table =
self.cat.get_table('default.split_reset_t').copy(reset_options)
+ native_table = normal_table.copy({'scan.native-plan.enabled': 'true'})
+
+ normal = normal_table.new_read_builder().new_scan().plan()
+ native = native_table.new_read_builder().explain()
+
+ self.assertTrue(native.native_planned)
+ self.assertEqual(native.split_count, len(normal.splits()))
+ self.assertEqual(native.split_count, 1)
+
+ def test_partitioned_table_falls_back(self):
+ # Rust bucket_path can diverge from the writer's str(value) partition
dir -> fall back.
+ schema = pa.schema([('k', pa.int64()), ('p', pa.string())])
+ self.cat.create_table('default.pt_t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['p']), False)
+ t = self.cat.get_table('default.pt_t')
+ wb = t.new_batch_write_builder()
+ w, c = wb.new_write(), wb.new_commit()
+ w.write_arrow(pa.Table.from_pylist(
+ [{'k': 1, 'p': 'a'}, {'k': 2, 'p': 'a'}, {'k': 3, 'p': 'b'}],
schema=schema))
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ native_table = self.cat.get_table('default.pt_t').copy(
+ {'scan.native-plan.enabled': 'true'})
+
self.assertFalse(native_table.new_read_builder().explain().native_planned)
+
+ def test_explain_reflects_native_plan(self):
+ self.cat.create_table(
+ 'default.ex_t', Schema.from_pyarrow_schema(self.schema), False)
+ self._write('ex_t', [{'k': 1, 'v': 'a'}, {'k': 2, 'v': 'b'}])
+ normal =
self.cat.get_table('default.ex_t').new_read_builder().explain()
+ native = self.cat.get_table('default.ex_t').copy(
+ {'scan.native-plan.enabled': 'true'}).new_read_builder().explain()
+ self.assertFalse(normal.native_planned)
+ self.assertTrue(native.native_planned)
+ self.assertEqual(native.split_count, normal.split_count)
+ self.assertEqual(native.snapshot_id, normal.snapshot_id)
+ self.assertIn('native', str(native)) # render shows the Planner line
+
+ def test_empty_table_explain_reflects_python_fallback(self):
+ self.cat.create_table(
+ 'default.empty_t', Schema.from_pyarrow_schema(self.schema), False)
+ normal =
self.cat.get_table('default.empty_t').new_read_builder().explain()
+ fallback = self.cat.get_table('default.empty_t').copy(
+ {'scan.native-plan.enabled': 'true'}).new_read_builder().explain()
+
+ self.assertFalse(fallback.native_planned)
+ self.assertEqual(fallback.snapshot_id, normal.snapshot_id)
+ self.assertEqual(fallback.split_count, 0)
+ self.assertNotIn('Planner:', str(fallback))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
new file mode 100644
index 0000000000..b21ac808a6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -0,0 +1,379 @@
+# 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 sys
+import unittest
+from types import ModuleType
+from unittest.mock import Mock, patch
+
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.catalog.filesystem_catalog_loader import FileSystemCatalogLoader
+from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader
+from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
+from pypaimon.read.native_plan import _catalog_options, native_plan
+from pypaimon.read.scan_stats import ScanStats
+from pypaimon.read.table_scan import TableScan
+from pypaimon.table.bucket_mode import BucketMode
+
+
+def _scan(native_enabled, file_scanner):
+ """Build a TableScan without running its heavy __init__."""
+ scan = TableScan.__new__(TableScan)
+ scan.table = Mock()
+ scan.table.options.native_plan_enabled.return_value = native_enabled
+ scan.table.options.options.contains_key.return_value = False # no
time-travel
+ scan.table.options.options.contains.return_value = False # no
incremental
+ scan.table.options.merge_engine.return_value = None # not
first-row
+ scan.table.options.query_auth_enabled = False
+ scan.table.current_branch.return_value = 'main'
+ scan.table.is_primary_key_table = False # not a pk table
+ scan.table.trimmed_primary_keys = ['k'] # non-empty trimmed pk
+ scan.table.bucket_mode.return_value = BucketMode.HASH_FIXED
+ scan.table._applied_dynamic_options = {} # no copy() overrides
+ scan.table.partition_keys = [] # not partitioned
+ scan.table.table_schema.id = 1
+ scan.table.schema_manager.latest.return_value.id = 1 # loaded schema is
latest
+ scan.table.identifier.get_database_name.return_value = 'default'
+ scan.table.catalog_environment.catalog_loader = FileSystemCatalogLoader(
+ CatalogContext.create_from_options(Options({}))) #
filesystem catalog
+ file_scanner.idx_of_this_subtask = None # no shard
+ file_scanner.start_pos_of_this_subtask = None # no slice
+ file_scanner.chunk_shuffle = None # no chunk-shuffle
+ file_scanner._global_index_result = None # no global-index result
+ file_scanner.deletion_vectors_enabled = False # no deletion vectors
+ file_scanner.data_evolution = False # no data evolution
+ file_scanner.only_read_real_buckets = False # not postpone bucket
+ scan.file_scanner = file_scanner
+ scan._query_auth_fn = None # no query-auth restrictions
+ scan._read_type = None
+ scan.limit = None # no row limit
+ return scan
+
+
+class NativePlanTest(unittest.TestCase):
+
+ def setUp(self):
+ # Make the real capability probe see a split-API-capable pypaimon-rust
so
+ # gate tests route natively. Tests that call native_plan() directly
override
+ # sys.modules within their own block.
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.PaimonCatalog = type(
+ 'PaimonCatalog', (), {'get_table': lambda self, name: None})
+ fake_df.Split = type('Split', (), {'serialize': lambda self: b''})
+ fake_mod = ModuleType('pypaimon_rust')
+ fake_mod.datafusion = fake_df
+ patcher = patch.dict(
+ sys.modules,
+ {'pypaimon_rust': fake_mod, 'pypaimon_rust.datafusion': fake_df})
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ 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_plan_uses_file_scanner_when_switch_off(self):
+ fs = Mock()
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=False, file_scanner=fs)
+ self.assertIs(scan.plan(), sentinel)
+ fs.scan.assert_called_once_with()
+
+ def test_plan_routes_to_native_and_prunes_partitions(self):
+ # Native planner returns every partition; the predicate keeps only
[2026, 7].
+ keep = Mock(partition=Mock(values=[2026, 7]))
+ drop = Mock(partition=Mock(values=[2025, 1]))
+ pred = Mock()
+ pred.test.side_effect = lambda part: part.values == [2026, 7]
+ fs = Mock(partition_key_predicate=pred)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+
+ with patch('pypaimon.read.native_plan.native_plan',
return_value=[keep, drop]) as np:
+ plan = scan.plan()
+
+ np.assert_called_once_with(scan.table)
+ fs.scan.assert_not_called()
+ self.assertEqual(plan.splits(), [keep])
+
+ def test_plan_falls_back_when_partition_prune_raises(self):
+ pred = Mock()
+ pred.test.side_effect = RuntimeError('predicate boom')
+ fs = Mock(partition_key_predicate=pred)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ split = Mock(partition=Mock(values=[2026, 7]), snapshot_id=1)
+ with patch('pypaimon.read.native_plan.native_plan',
return_value=[split]):
+ self.assertIs(scan.plan(), sentinel)
+ fs.scan.assert_called_once_with()
+
+ def test_plan_native_no_partition_predicate_keeps_all(self):
+ splits = [Mock(partition=Mock(values=[1])),
Mock(partition=Mock(values=[2]))]
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ with patch('pypaimon.read.native_plan.native_plan',
return_value=splits):
+ self.assertEqual(scan.plan().splits(), splits)
+
+ def test_plan_falls_back_when_scan_is_not_plain(self):
+ # Native planning does not carry shard/slice, global-index, or
+ # time-travel/incremental scans -> must fall back to the file scanner.
+ def check(setup):
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ setup(scan, fs)
+ with patch('pypaimon.read.native_plan.native_plan') as np:
+ self.assertIs(scan.plan(), sentinel)
+ np.assert_not_called()
+ fs.scan.assert_called_once_with()
+
+ check(lambda s, fs: setattr(fs, 'idx_of_this_subtask', 0))
+ check(lambda s, fs: 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, 'deletion_vectors_enabled', True))
+ check(lambda s, fs: setattr(fs, 'data_evolution', True))
+ check(lambda s, fs: setattr(fs, 'only_read_real_buckets', True))
+ check(lambda s, fs: (setattr(s.table, 'is_primary_key_table', True),
+ setattr(s.table, 'trimmed_primary_keys', [])))
+ check(lambda s, fs: s.table.bucket_mode.__setattr__(
+ 'return_value', BucketMode.HASH_DYNAMIC))
+ check(lambda s, fs: s.table.bucket_mode.__setattr__(
+ 'return_value', BucketMode.CROSS_PARTITION))
+ check(lambda s, fs: setattr(
+ s.table, '_applied_dynamic_options', {'scan.snapshot-id': None}))
+ check(lambda s, fs: setattr(s.table, 'partition_keys', ['dt']))
+ 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: setattr(s, 'limit', 5))
+ check(lambda s, fs: s.table.options.options.contains_key.__setattr__(
+ 'side_effect', lambda k: k == 'scan.version'))
+ 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__(
+ 'return_value', 'db.name'))
+ check(lambda s, fs: s.table.identifier.get_database_name.__setattr__(
+ 'return_value', 'unknown'))
+ check(lambda s, fs: setattr(
+ s.table.catalog_environment, 'catalog_loader', object())) # no
context()
+ for attr in ('hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'):
+ check(lambda s, fs, attr=attr: setattr(
+ s.table.catalog_environment.catalog_loader.context(), attr,
object()))
+ check(lambda s, fs: s.table.options.options.contains_key.__setattr__(
+ 'return_value', True)) # time-travel
+ check(lambda s, fs: s.table.options.options.contains.__setattr__(
+ 'return_value', True)) # incremental
+
+ def test_plan_native_empty_falls_back(self):
+ # Empty native result -> fall back for an atomic snapshot id.
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ with patch('pypaimon.read.native_plan.native_plan', return_value=[]):
+ self.assertIs(scan.plan(), sentinel)
+ fs.scan.assert_called_once_with()
+
+ def test_plan_falls_back_when_rust_unavailable(self):
+ # scan.native-plan.enabled but pypaimon-rust missing/old -> fall back,
+ # not crash.
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ with patch('pypaimon.read.native_plan.native_runtime_available',
+ return_value=False), \
+ patch('pypaimon.read.native_plan.native_plan') as np:
+ self.assertIs(scan.plan(), sentinel)
+ np.assert_not_called()
+ fs.scan.assert_called_once_with()
+
+ def test_plan_falls_back_when_native_plan_raises(self):
+ # A native planning failure (e.g. unsupported scheme) must fall back,
not crash.
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ with patch('pypaimon.read.native_plan.native_plan',
+ side_effect=RuntimeError('unsupported scheme viewfs://')):
+ self.assertIs(scan.plan(), sentinel)
+ fs.scan.assert_called_once_with()
+
+ def test_plan_falls_back_for_jdbc_catalog_loader(self):
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ scan.table.catalog_environment.catalog_loader = JdbcCatalogLoader(
+ CatalogContext.create_from_options(Options({})))
+
+ with patch('pypaimon.read.native_plan.native_plan') as np:
+ self.assertIs(scan.plan(), sentinel)
+
+ np.assert_not_called()
+ fs.scan.assert_called_once_with()
+
+ def test_plan_falls_back_for_builtin_catalog_loader_subclasses(self):
+ class RoutedFileSystemLoader(FileSystemCatalogLoader):
+ def load(self):
+ return object()
+
+ class RoutedRESTLoader(RESTCatalogLoader):
+ def load(self):
+ return object()
+
+ for loader_class in (RoutedFileSystemLoader, RoutedRESTLoader):
+ with self.subTest(loader_class=loader_class.__name__):
+ fs = Mock(partition_key_predicate=None)
+ sentinel = object()
+ fs.scan.return_value = sentinel
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ scan.table.catalog_environment.catalog_loader = loader_class(
+ CatalogContext.create_from_options(Options({})))
+
+ with patch('pypaimon.read.native_plan.native_plan') as np:
+ self.assertIs(scan.plan(), sentinel)
+
+ np.assert_not_called()
+ fs.scan.assert_called_once_with()
+
+ def test_scan_with_stats_native_empty_uses_fallback_stats(self):
+ fs = Mock(partition_key_predicate=None)
+ fallback_plan = object()
+ fallback_stats = ScanStats(manifest_files_total=7)
+ fs.scan_with_stats.return_value = (fallback_plan, fallback_stats)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+
+ with patch('pypaimon.read.native_plan.native_plan', return_value=[])
as np:
+ plan, stats = scan.scan_with_stats()
+
+ self.assertIs(plan, fallback_plan)
+ self.assertIs(stats, fallback_stats)
+ np.assert_called_once_with(scan.table)
+ fs.scan_with_stats.assert_called_once_with()
+ fs.scan.assert_not_called()
+
+ def test_catalog_options_are_normalized_for_rust(self):
+ table = Mock()
+ table.catalog_environment.catalog_loader = FileSystemCatalogLoader(
+ CatalogContext.create_from_options(Options({
+ 'warehouse': '/tmp/warehouse',
+ 'data-token.enabled': True,
+ 'retry-count': 3,
+ 'unset': None,
+ })))
+
+ self.assertEqual(_catalog_options(table), {
+ 'warehouse': '/tmp/warehouse',
+ # Lowercase so Rust's case-sensitive bool parser accepts it.
+ 'data-token.enabled': 'true',
+ 'retry-count': '3',
+ 'metastore': 'filesystem',
+ })
+
+ def test_catalog_options_use_actual_rest_loader_type(self):
+ table = Mock()
+ table.catalog_environment.catalog_loader = RESTCatalogLoader(
+ CatalogContext.create_from_options(Options({
+ 'uri': 'http://localhost:8181',
+ })))
+
+ self.assertEqual(_catalog_options(table), {
+ 'uri': 'http://localhost:8181',
+ 'metastore': 'rest',
+ })
+
+ def test_catalog_options_reject_loader_subclass(self):
+ class RoutedFileSystemLoader(FileSystemCatalogLoader):
+ pass
+
+ table = Mock()
+ table.catalog_environment.catalog_loader = RoutedFileSystemLoader(
+ CatalogContext.create_from_options(Options({})))
+
+ with self.assertRaisesRegex(ValueError, 'exact built-in catalog
loader'):
+ _catalog_options(table)
+
+ def test_native_plan_threads_trimmed_keys_to_deserializer(self):
+ # PK tables route through: the trimmed primary keys must reach the
+ # deserializer so per-file min/max keys are decoded for merge-on-read.
+ kfields = [object()]
+ table = Mock(trimmed_primary_keys_fields=kfields)
+ table.table_schema = Mock(fields=[], partition_keys=[])
+ table.options.source_split_target_size.return_value = 1024
+ table.options.source_split_open_file_cost.return_value = 128
+ split = Mock()
+ split.serialize.return_value = b'bytes'
+ rt = Mock()
+
rt.new_read_builder.return_value.new_scan.return_value.plan.return_value \
+ .splits.return_value = [split]
+ catalog = Mock()
+ catalog.get_table.return_value = rt
+
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.PaimonCatalog = Mock(return_value=catalog)
+ fake_df.Split = type('Split', (), {'serialize': lambda self: b''})
+ fake_mod = ModuleType('pypaimon_rust')
+ fake_mod.datafusion = fake_df
+
+ with patch.dict(sys.modules,
+ {'pypaimon_rust': fake_mod,
'pypaimon_rust.datafusion': fake_df}), \
+ patch('pypaimon.read.native_plan._catalog_options',
return_value={}), \
+ patch('pypaimon.read.native_plan.deserialize_split_v1',
+ return_value='decoded') as des:
+ result = native_plan(table)
+
+ self.assertEqual(result, ['decoded'])
+ rt.new_read_builder.assert_called_once_with({
+ CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): '1024',
+ CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): '128',
+ })
+ des.assert_called_once_with(b'bytes', [], kfields)
+
+ def test_native_plan_requires_split_api(self):
+ # An intermediate pypaimon-rust missing either get_table or
Split.serialize
+ # must raise a clear error, not an AttributeError mid-plan.
+ split_ok = type('Split', (), {'serialize': lambda self: b''})
+ catalog_ok = type('PaimonCatalog', (), {'get_table': lambda self,
name: None})
+ cases = {
+ 'no get_table': (type('PaimonCatalog', (), {}), split_ok),
+ 'no serialize': (catalog_ok, type('Split', (), {})),
+ }
+ for label, (catalog_cls, split_cls) in cases.items():
+ with self.subTest(case=label):
+ fake_df = ModuleType('pypaimon_rust.datafusion')
+ fake_df.PaimonCatalog = catalog_cls
+ fake_df.Split = split_cls
+ fake_mod = ModuleType('pypaimon_rust')
+ fake_mod.datafusion = fake_df
+ with patch.dict(
+ sys.modules,
+ {'pypaimon_rust': fake_mod,
'pypaimon_rust.datafusion': fake_df}):
+ with self.assertRaisesRegex(RuntimeError, '0.3.0'):
+ native_plan(Mock())
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/split_serializer_test.py
b/paimon-python/pypaimon/tests/split_serializer_test.py
new file mode 100644
index 0000000000..d168c6ff4b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/split_serializer_test.py
@@ -0,0 +1,170 @@
+# 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 base64
+import unittest
+
+from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.read.split_serializer import (
+ _decode_modified_utf8, _decode_str_array, deserialize_split_v1)
+from pypaimon.schema.data_types import AtomicType, DataField
+
+# Hand-built BinaryArray<string> == ["id", "longcolumn12"]: hits both the
inline
+# (<=7 bytes) and var-pointer (>7) element encodings.
+_BINARY_ARRAY_STR = (
+ bytes([0x02, 0x00, 0x00, 0x00, # n = 2
+ 0x00, 0x00, 0x00, 0x00, # null bitset
+ 0x69, 0x64, 0x00, 0x00, 0x00, 0x00, 0x00, 0x82, # "id" inline (len
2)
+ 0x0C, 0x00, 0x00, 0x00, 0x18, 0x00, 0x00, 0x00]) # ptr len=12
off=24
+ + b"longcolumn12" + bytes(4))
+
+# Golden SplitSerializer v1: DataSplit, partition [2026, 7], bucket 3,
+# bucket_path "dt=20260706/bucket-3", files file-a/file-b, dv on file-b.
+# Captured from the Java SplitSerializer (the reference wire format); live
+# pypaimon_rust byte-compat is covered by native_plan_integration_test.
+_GOLDEN_DATA_SPLIT_V1 = base64.b64decode(
+
"U1BMSVRfVjEAAAABAAAAAd7D0jAsGexmAAAACAAAAAAAAAAqAAAAHAAAAAIAAAAAAAAAAOoHAAAA"
+
"AAAABwAAAAAAAAAAAAADABRkdD0yMDI2MDcwNi9idWNrZXQtMwEAAAAIAAAAAAAAAAACAAABcAAA"
+
"QA8AAAAAZmlsZS1hAIYKAAAAAAAAAAoAAAAAAAAAFAAAAKgAAAAUAAAAwAAAAEgAAADYAAAASAAA"
+
"ACABAAAAAAAAAAAAAGQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAgAAABoAQAAZAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEA"
+
"AAAAAAAAAAEAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAoAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAA"
+
"AAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAADAAAACAAAAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABcAAAQA8AAAAAZmlsZS1iAIYKAAAAAAAAAAoAAAAA"
+
"AAAAFAAAAKgAAAAUAAAAwAAAAEgAAADYAAAASAAAACABAAAAAAAAAAAAAMgAAAAAAAAAAAAAAAAA"
+
"AAABAAAAAAAAAAgAAABoAQAAZAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAsAAAAAAAAAAAAAAAAAAAEAAAAA"
+
"AAAAABQAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAAAAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAAAAAMAAAAMAAAAAgA"
+
"AABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQAA"
+ "AAIAAQAJZHYvZmlsZS1iAAAAAAAAAAIAAAAAAAAACgAAAAAAAAADAAE="
+)
+
+# Same split serialized as an IndexedSplit (type id 3): identical DataSplit
body
+# with a trailing row-ranges/scores section that the reader skips.
+_GOLDEN_INDEXED_SPLIT_V1 = base64.b64decode(
+
"U1BMSVRfVjEAAAABAAAAA/L54FRCJC4xAAAAAd7D0jAsGexmAAAACAAAAAAAAAAqAAAAHAAAAAIA"
+
"AAAAAAAAAOoHAAAAAAAABwAAAAAAAAAAAAADABRkdD0yMDI2MDcwNi9idWNrZXQtMwEAAAAIAAAA"
+
"AAAAAAACAAABcAAAQA8AAAAAZmlsZS1hAIYKAAAAAAAAAAoAAAAAAAAAFAAAAKgAAAAUAAAAwAAA"
+
"AEgAAADYAAAASAAAACABAAAAAAAAAAAAAGQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAgAAABoAQAA"
+
"ZAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAEAAAAAAAAAAAEAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAoAAAAAAAAAAAAAAAAA"
+
"AAAAAAAADAAAACAAAAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAAAAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABcAAAQA8AAAAAZmlsZS1iAIYK"
+
"AAAAAAAAAAoAAAAAAAAAFAAAAKgAAAAUAAAAwAAAAEgAAADYAAAASAAAACABAAAAAAAAAAAAAMgA"
+
"AAAAAAAAAAAAAAAAAAABAAAAAAAAAAgAAABoAQAAZAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAsAAAAAAAAA"
+
"AAAAAAAAAAEAAAAAAAAAABQAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAAAAAMAAAAMAAAAAgAAABA"
+
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADAAAACAA"
+
"AAAMAAAAMAAAAAgAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAAAAAAAAAAQAAAAIAAQAJZHYvZmlsZS1iAAAAAAAAAAIAAAAAAAAACgAAAAAAAAADAAEAAAAC"
+ "AAAAAAAAAAEAAAAAAAAABAAAAAAAAAALAAAAAAAAAA0BAAAAAz8AAAA+gAAAPgAAAA=="
+)
+
+
+class SplitSerializerTest(unittest.TestCase):
+
+ def _partition_fields(self):
+ return [DataField(0, 'y', AtomicType('INT')),
+ DataField(1, 'm', AtomicType('INT'))]
+
+ def test_deserialize_data_split_v1_golden(self):
+ split = deserialize_split_v1(_GOLDEN_DATA_SPLIT_V1,
self._partition_fields())
+
+ self.assertIsNotNone(split.snapshot_id)
+ self.assertEqual(split.bucket, 3)
+ self.assertEqual(list(split.partition.values), [2026, 7])
+
+ self.assertEqual([f.file_name for f in split.files], ['file-a',
'file-b'])
+ self.assertEqual(
+ [f.file_path for f in split.files],
+ ['dt=20260706/bucket-3/file-a', 'dt=20260706/bucket-3/file-b'])
+ self.assertEqual([f.level for f in split.files], [0, 1])
+ self.assertEqual(split.files[0].max_sequence_number, 100)
+ self.assertEqual(split.files[1].max_sequence_number, 200)
+
+ self.assertEqual(len(split.data_deletion_files), 2)
+ self.assertIsNone(split.data_deletion_files[0])
+ dv = split.data_deletion_files[1]
+ self.assertEqual(
+ (dv.dv_index_path, dv.offset, dv.length, dv.cardinality),
+ ('dv/file-b', 2, 10, 3))
+
+ def test_decodes_min_max_keys_with_key_fields(self):
+ # Trimmed primary keys -> per-file min/max keys are decoded for PK
+ # merge-on-read. The golden files carry keys [1..10] and [11..20].
+ key_fields = [DataField(0, 'k', AtomicType('BIGINT'))]
+ split = deserialize_split_v1(
+ _GOLDEN_DATA_SPLIT_V1, self._partition_fields(), key_fields)
+ self.assertEqual([list(f.min_key.values) for f in split.files], [[1],
[11]])
+ self.assertEqual([list(f.max_key.values) for f in split.files], [[10],
[20]])
+
+ def test_keys_stay_empty_without_key_fields(self):
+ # Append tables pass no key fields; keys stay empty (unchanged
behavior).
+ split = deserialize_split_v1(_GOLDEN_DATA_SPLIT_V1,
self._partition_fields())
+ self.assertEqual([list(f.min_key.values) for f in split.files], [[],
[]])
+ self.assertEqual([list(f.max_key.values) for f in split.files], [[],
[]])
+
+ def test_deserialize_indexed_split_v1_golden(self):
+ # type id 3 -> IndexedSplit; row_ranges/scores must be preserved
(dropping
+ # them would make the reader scan the whole file, not the ANN/row-id
result).
+ split = deserialize_split_v1(_GOLDEN_INDEXED_SPLIT_V1,
self._partition_fields())
+ self.assertIsInstance(split, IndexedSplit)
+ self.assertIsNotNone(split.snapshot_id) # delegates to the inner
DataSplit
+ self.assertEqual(split.bucket, 3)
+ self.assertEqual([f.file_name for f in split.files], ['file-a',
'file-b'])
+ self.assertEqual(
+ [(r.from_, r.to) for r in split.row_ranges()], [(1, 4), (11, 13)])
+ self.assertEqual(split.scores(), [0.5, 0.25, 0.125])
+
+ def test_decode_modified_utf8_supplementary_char(self):
+ # Java writeUTF encodes U+1F600 as a CESU-8 surrogate pair; must
recombine.
+ self.assertEqual(
+ _decode_modified_utf8(bytes([0xED, 0xA0, 0xBD, 0xED, 0xB8, 0x80])),
+ '\U0001F600')
+
+ def test_decode_str_array_inline_and_pointer(self):
+ # Covers both element encodings: inline (<=7 bytes) and var pointer
(>7).
+ self.assertEqual(_decode_str_array(_BINARY_ARRAY_STR), ['id',
'longcolumn12'])
+ self.assertIsNone(_decode_str_array(None))
+ # Empty array: count 0 + empty null bitset.
+ self.assertEqual(_decode_str_array(bytes([0, 0, 0, 0])), [])
+
+ def test_bad_magic_and_version_raise(self):
+ good = _GOLDEN_DATA_SPLIT_V1
+ pf = self._partition_fields()
+ with self.assertRaisesRegex(ValueError, "magic"):
+ deserialize_split_v1(b'\x00' * 8 + good[8:], pf)
+ bad_version = good[:8] + (99).to_bytes(4, 'big') + good[12:]
+ with self.assertRaisesRegex(ValueError, "version"):
+ deserialize_split_v1(bad_version, pf)
+
+ def test_bad_indexed_magic_and_version_raise(self):
+ # IndexedSplit magic/version follow the 16-byte SplitSerializer header.
+ good = _GOLDEN_INDEXED_SPLIT_V1
+ pf = self._partition_fields()
+ with self.assertRaisesRegex(ValueError, "IndexedSplit magic"):
+ deserialize_split_v1(good[:16] + b'\x00' * 8 + good[24:], pf)
+ bad_version = good[:24] + (99).to_bytes(4, 'big') + good[28:]
+ with self.assertRaisesRegex(ValueError, "IndexedSplit version"):
+ deserialize_split_v1(bad_version, pf)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/setup.py b/paimon-python/setup.py
index f66717fb91..2fa5a3be7d 100644
--- a/paimon-python/setup.py
+++ b/paimon-python/setup.py
@@ -184,7 +184,7 @@ setup(
'paimon-vindex==0.1.0; python_version>="3.9"',
],
'sql': [
- 'pypaimon-rust; python_version>="3.10"',
+ 'pypaimon-rust>=0.3.0,<0.4.0; python_version>="3.10"',
'datafusion>=52; python_version>="3.10"',
],
'hdfs': [
