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 43e9ad701e [python] Run native planning CI against paimon-rust main
(#9817)
43e9ad701e is described below
commit 43e9ad701ebee6d9af8316f35533f4d0836a1fc0
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 14 19:22:09 2026 +0800
[python] Run native planning CI against paimon-rust main (#9817)
---
.github/workflows/ci-python.yml | 77 +++++++++++++++++-----
paimon-python/pypaimon/read/split_serializer.py | 24 +++----
.../tests/manifest/manifest_manager_test.py | 25 ++++---
.../pypaimon/tests/pushdown_bucket_test.py | 2 +
.../pypaimon/tests/split_serializer_test.py | 68 ++++++++++++++++++-
5 files changed, 158 insertions(+), 38 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index daffd050ad..a33b09820e 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -28,7 +28,6 @@ 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: be5456e96b326939570b6ef5d63acddda3311c4f
jobs:
@@ -90,12 +89,6 @@ jobs:
java -version
mvn -version
- - name: Install Rust toolchain
- if: matrix.python-version == '3.11'
- run: |
- curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s --
-y --default-toolchain stable --profile minimal
- echo "$HOME/.cargo/bin" >> $GITHUB_PATH
-
- name: Verify Python version
run: python --version
@@ -155,14 +148,10 @@ jobs:
python -c "import datasets, lerobot; print('datasets',
datasets.__version__, 'lerobot', lerobot.__version__)"
fi
+ python -m pip install pypaimon-rust
if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
# Run the RoboMIND pipeline tests with synthetic local HDF5 data.
- # Exercise the 0.4 API in one lane until its wheel is published.
- python -m pip install
"git+https://github.com/apache/paimon-rust.git@${PYPAIMON_RUST_REV}#subdirectory=bindings/python";
python -m pip install "./paimon-python[sql]"
- python -c "import importlib.metadata as m; from
pypaimon_rust.datafusion import PaimonCatalog, Split; assert
m.version('pypaimon-rust') == '0.4.0'; assert hasattr(PaimonCatalog,
'get_table'); assert hasattr(Split, 'serialize')"
- else
- python -m pip install pypaimon-rust==0.3.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
@@ -180,11 +169,69 @@ jobs:
shell: bash
run: |
chmod +x paimon-python/dev/lint-python.sh
- if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
- export PYPAIMON_TEST_NATIVE_PLAN=1
- fi
./paimon-python/dev/lint-python.sh -e pytest_torch
+ rust-plan:
+ name: Rust Plan
+ timeout-minutes: 90
+ runs-on: ubuntu-latest
+ container: "python:3.11-slim"
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v6
+
+ - name: Install system dependencies
+ run: |
+ apt-get update && apt-get install -y --no-install-recommends \
+ build-essential git curl cmake pkg-config libssl-dev libclang-dev \
+ && rm -rf /var/lib/apt/lists/*
+
+ - name: Install Rust toolchain
+ run: |
+ curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s --
-y --default-toolchain stable --profile minimal
+ echo "$HOME/.cargo/bin" >> $GITHUB_PATH
+
+ - name: Build paimon-rust main
+ run: |
+ python -m pip install --upgrade pip
+ python -m pip install --upgrade
"git+https://github.com/apache/paimon-rust.git@main#subdirectory=bindings/python";
+
+ - name: Install Python test dependencies
+ run: |
+ python -m pip install torch --index-url
https://download.pytorch.org/whl/cpu
+ python -m pip install "./paimon-python[sql,rosbag]" \
+ readerwriterlock==1.0.9 fsspec==2024.3.1 cachetools==5.3.3
ossfs==2023.12.0 \
+ ray==2.54.0 fastavro==1.11.1 zstandard==0.24.0 polars==1.32.0
duckdb==1.3.2 \
+ pylance==0.39.0 'pytest~=7.0' py4j==0.10.9.9 parameterized==0.9.0 \
+ 'daft>=0.7.6' datasketches pyarrow==18.1.0 numpy==1.24.3
pandas==2.0.3 \
+ 'h5py>=3,<4' 'lumina-data>=${{ env.LUMINA_DATA_VERSION }}' \
+ vortex-data==0.70.0 'paimon-mosaic>=0.1.0'
+ python -m pip check
+
+ - name: Verify upstream native planning APIs
+ run: |
+ python - <<'PY'
+ import importlib.metadata as metadata
+ import json
+
+ from pypaimon_rust.datafusion import PaimonCatalog, ReadBuilder,
Split
+
+ distribution = metadata.distribution('pypaimon-rust')
+ source = json.loads(distribution.read_text('direct_url.json'))
+ print('pypaimon-rust', distribution.version,
+ 'commit', source['vcs_info']['commit_id'])
+ assert hasattr(PaimonCatalog, 'get_table'), 'Missing
PaimonCatalog.get_table'
+ assert hasattr(Split, 'serialize'), 'Missing Split.serialize'
+ assert hasattr(ReadBuilder, 'with_row_ranges'), 'Missing
ReadBuilder.with_row_ranges'
+ PY
+
+ - name: Run Python tests with Rust planning
+ shell: bash
+ env:
+ PYPAIMON_TEST_NATIVE_PLAN: '1'
+ run: |
+ bash paimon-python/dev/lint-python.sh -i pytest
+
torch:
name: PyTorch / Python 3.10
timeout-minutes: 60
diff --git a/paimon-python/pypaimon/read/split_serializer.py
b/paimon-python/pypaimon/read/split_serializer.py
index 4b50c86c28..ff5397c7c2 100644
--- a/paimon-python/pypaimon/read/split_serializer.py
+++ b/paimon-python/pypaimon/read/split_serializer.py
@@ -18,7 +18,7 @@
"""Deserialize the cross-language ``SplitSerializer`` v1 binary into a pypaimon
:class:`DataSplit`.
-Mirror of the Java ``DataSplit#serialize`` (VERSION 8) frame wrapped in the
+Mirror of the Java ``DataSplit#serialize`` (versions 8 and 9) 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.
@@ -46,16 +46,14 @@ _TYPE_INDEXED_SPLIT = 3
_INDEXED_SPLIT_MAGIC = -938472394838495695
_INDEXED_SPLIT_VERSION = 1
_SPLIT_MAGIC = -2394839472490812314
-_SPLIT_VERSION = 8
-
-_DFM_ARITY = 20
+_DFM_ARITY_BY_VERSION = {8: 20, 9: 21}
def _f(idx, name, dtype):
return DataField(idx, name, dtype)
-# DataFileMeta 20-field layout (order/types mirror
DataFileMetaSerializer#toRow).
+# DataFileMeta 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')),
@@ -78,10 +76,11 @@ _DFM_FIELDS: List[DataField] = [
_f(17, '_EXTERNAL_PATH', AtomicType('STRING')),
_f(18, '_FIRST_ROW_ID', AtomicType('BIGINT')),
_f(19, '_WRITE_COLS', AtomicType('BYTES')),
+ # Added in v9 for compaction/index refresh; unused by the Python reader.
+ _f(20, '_WRITE_COLS_SEQUENCES', AtomicType('BYTES')),
]
-# Arity is fixed by DataSplit VERSION 8; keep the field list and arity in
lockstep.
-assert len(_DFM_FIELDS) == _DFM_ARITY
+assert len(_DFM_FIELDS) == max(_DFM_ARITY_BY_VERSION.values())
def _decode_str_array(b: Optional[bytes]) -> Optional[List[str]]:
@@ -214,9 +213,9 @@ def _read_datasplit_body(r: _Reader, partition_fields:
List[DataField],
if r.i64() != _SPLIT_MAGIC:
raise ValueError("bad DataSplit magic")
version = r.i32()
- if version != _SPLIT_VERSION:
+ if version not in _DFM_ARITY_BY_VERSION:
raise ValueError(
- "unsupported DataSplit version %d (expected %d)" % (version,
_SPLIT_VERSION))
+ "unsupported DataSplit version %d (expected 8 or 9)" % 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()
@@ -228,7 +227,8 @@ def _read_datasplit_body(r: _Reader, partition_fields:
List[DataField],
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)
+ arity = _DFM_ARITY_BY_VERSION[version]
+ files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, arity,
key_fields)
for _ in range(file_count)]
data_deletion_files = _read_deletion_list(r)
r.u8() # isStreaming
@@ -250,9 +250,9 @@ def _decode_key(b: Optional[bytes], key_fields:
Optional[List[DataField]]) -> Ge
return GenericRowDeserializer.from_bytes(b, key_fields)
-def _datafilemeta_from_row(row_bytes: bytes, bucket_path: str,
+def _datafilemeta_from_row(row_bytes: bytes, bucket_path: str, arity: int,
key_fields: Optional[List[DataField]] = None) ->
DataFileMeta:
- row = BinaryRow(struct.pack('>i', _DFM_ARITY) + row_bytes, _DFM_FIELDS)
+ row = BinaryRow(struct.pack('>i', arity) + row_bytes, _DFM_FIELDS[:arity])
g = row.get_field
file_name = g(0)
external_path = g(17)
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
index 65be8ebdc5..f1def7a193 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
@@ -52,7 +52,7 @@ def _paimon_python_root():
def _runner_can_write_zstandard_avro():
- """fastavro uses ``backports.zstd`` (Py < 3.14) to *write* zstandard Avro
blocks."""
+ """Whether fastavro can write zstandard Avro blocks with the installed
backends."""
try:
from io import BytesIO
import fastavro
@@ -89,8 +89,8 @@ _MANIFEST_ZSTD_READ_SUBPROC_VENV_PYTHON = None
def _manifest_zstd_read_subprocess_venv_python():
"""Disposable venv with editable pypaimon for
``manifest_list_zstd_read_subprocess.py``.
- Does not install ``backports.zstd`` so the first worker run can hit
fastavro's missing zstd
- codec path when reading zstandard-compressed manifest lists.
+ The test removes the zstd backends before the first worker run, then
installs them
+ again to verify reading zstandard-compressed manifest lists in a fresh
process.
"""
global _MANIFEST_ZSTD_READ_SUBPROC_VENV_DIR,
_MANIFEST_ZSTD_READ_SUBPROC_VENV_PYTHON
with _MANIFEST_ZSTD_READ_SUBPROC_VENV_LOCK:
@@ -99,18 +99,21 @@ def _manifest_zstd_read_subprocess_venv_python():
repo = _paimon_python_root()
venv_dir = tempfile.mkdtemp(prefix='paimon-zstd-read-subprocess-')
pip_install_env = _subprocess_env_for_pip()
+ # Keep the codec behavior identical to the parent test environment.
+ fastavro_requirement = 'fastavro=={}'.format(fastavro.__version__)
uv_bin = shutil.which('uv')
try:
if uv_bin:
subprocess.check_call(
- [uv_bin, 'venv', venv_dir],
+ [uv_bin, 'venv', '--python', sys.executable, '--seed',
venv_dir],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=pip_install_env,
)
isolated_venv_python = _venv_python_executable(venv_dir)
subprocess.check_call(
- [uv_bin, 'pip', 'install', '-q', '--python',
isolated_venv_python, '-e', repo, 'requests'],
+ [uv_bin, 'pip', 'install', '-q', '--python',
isolated_venv_python,
+ '-e', repo, 'requests', fastavro_requirement],
env=pip_install_env,
)
else:
@@ -121,7 +124,8 @@ def _manifest_zstd_read_subprocess_venv_python():
)
isolated_venv_python = _venv_python_executable(venv_dir)
subprocess.check_call(
- [isolated_venv_python, '-m', 'pip', 'install', '-q', '-e',
repo, 'requests'],
+ [isolated_venv_python, '-m', 'pip', 'install', '-q', '-e',
repo,
+ 'requests', fastavro_requirement],
env=pip_install_env,
)
except Exception:
@@ -606,10 +610,10 @@ class ManifestListManagerTest(_ManifestManagerSetup):
sys.version_info >= (3, 13),
'fastavro >= 1.12 bundles zstd in compiled extension on Python 3.13+',
)
- def test_zstd_manifest_list_fastavro_requires_backports_zstd(self):
+ def test_zstd_manifest_list_requires_codec_backend(self):
"""Child venv runs ``manifest_list_zstd_read_subprocess`` (argv:
warehouse, table id, list file name).
- No ``backports.zstd`` in the venv → read fails; after ``pip install``
→ read succeeds.
+ No zstd backend in the venv → read fails; after ``pip install`` → read
succeeds.
"""
if not _runner_can_write_zstandard_avro():
self.skipTest('runner cannot write zstandard Avro')
@@ -629,7 +633,7 @@ class ManifestListManagerTest(_ManifestManagerSetup):
catalog_table_id = 'default.{}'.format(self._table_name)
isolated_venv_python = _manifest_zstd_read_subprocess_venv_python()
pip_install_env = _subprocess_env_for_pip()
- subprocess.run(
+ subprocess.check_call(
[isolated_venv_python, '-m', 'pip', 'uninstall', '-y',
'backports.zstd', 'zstandard'],
stdout=subprocess.DEVNULL,
@@ -654,7 +658,8 @@ class ManifestListManagerTest(_ManifestManagerSetup):
stderr_and_stdout = (
read_without_zstd_backend.stdout +
read_without_zstd_backend.stderr)
self.assertIn('zstandard codec is supported but you need to install',
stderr_and_stdout)
- self.assertIn('backports.zstd', stderr_and_stdout)
+ # fastavro 1.11 uses zstandard; newer versions use backports.zstd.
+ self.assertRegex(stderr_and_stdout, r"'(?:backports\.zstd|zstandard)'")
subprocess.check_call(
[isolated_venv_python, '-m', 'pip', 'install', '-q',
diff --git a/paimon-python/pypaimon/tests/pushdown_bucket_test.py
b/paimon-python/pypaimon/tests/pushdown_bucket_test.py
index 13292d6860..8f3c0814cc 100644
--- a/paimon-python/pypaimon/tests/pushdown_bucket_test.py
+++ b/paimon-python/pypaimon/tests/pushdown_bucket_test.py
@@ -759,7 +759,9 @@ class BucketPruningIntegrationTest(unittest.TestCase):
"Equal on PK still narrows to the writer's bucket "
"even when AND'd with a non-bucket-key predicate")
+ @pytest.mark.python_plan
def test_manifest_bucket_pruning_skips_file_reads(self):
+ # This test observes Python FileIO calls; Rust uses its own file I/O.
table = self._create_pk_table('manifest_bucket_pruning')
for i in range(self.NUM_BUCKETS):
self._write(table, [{'id': i, 'val': i * 11}])
diff --git a/paimon-python/pypaimon/tests/split_serializer_test.py
b/paimon-python/pypaimon/tests/split_serializer_test.py
index d168c6ff4b..0c35e10640 100644
--- a/paimon-python/pypaimon/tests/split_serializer_test.py
+++ b/paimon-python/pypaimon/tests/split_serializer_test.py
@@ -16,6 +16,7 @@
# under the License.
import base64
+import struct
import unittest
from pypaimon.globalindex.indexed_split import IndexedSplit
@@ -56,7 +57,7 @@ _GOLDEN_DATA_SPLIT_V1 = base64.b64decode(
)
# Same split serialized as an IndexedSplit (type id 3): identical DataSplit
body
-# with a trailing row-ranges/scores section that the reader skips.
+# with a trailing row-ranges/scores section that must survive decoding.
_GOLDEN_INDEXED_SPLIT_V1 = base64.b64decode(
"U1BMSVRfVjEAAAABAAAAA/L54FRCJC4xAAAAAd7D0jAsGexmAAAACAAAAAAAAAAqAAAAHAAAAAIA"
"AAAAAAAAAOoHAAAAAAAABwAAAAAAAAAAAAADABRkdD0yMDI2MDcwNi9idWNrZXQtMwEAAAAIAAAA"
@@ -78,6 +79,26 @@ _GOLDEN_INDEXED_SPLIT_V1 = base64.b64decode(
)
+# Java DataSplitCompatibleTest's v9 golden, also used by paimon-rust's
+# serialize_matches_datasplit_v9 test. Includes the new non-null per-column
+# sequence numbers [15, 100, 150, 200] after write_cols.
+_GOLDEN_DATA_SPLIT_V9 = base64.b64decode(
+
"3sPSMCwZ7GYAAAAJAAAAAAAAABIAAAAUAAAAAQAAAAAAAAAAYWFhYWEAAIUAAAAUAAdteSBwYXRo"
+
"AQAAACAAAAAAAAAAAAEAAAJoAAAAAAAAAABteV9maWxlhwAAEAAAAAAAAAQAAAAAAAAUAAAAsAAA"
+
"ABQAAADIAAAAYAAAAOAAAACAAAAAQAEAAA8AAAAAAAAAyAAAAAAAAAAFAAAAAAAAAAMAAAAAAAAA"
+
"GAAAAMABAABgTEpMfwEAAAsAAAAAAAAAAQIEAAAAAIMBAAAAAAAAACAAAADYAQAAGQAAAPgBAAAM"
+
"AAAAAAAAACgAAAAYAgAAKAAAAEACAAAAAAABAAAAAAAAAABtaW5fa2V5hwAAAAAAAAABAAAAAAAA"
+
"AABtYXhfa2V5hwAAAAAAAAAAAAAAABQAAAAgAAAAFAAAADgAAAAQAAAAUAAAAAAAAAEAAAAAAAAA"
+
"AG1pbl9rZXmHAAAAAAAAAAEAAAAAAAAAAG1heF9rZXmHAAAAAAEAAAAAAAAAAAAAAAAAAAAAAAAA"
+
"AAAAACQAAAAgAAAAJAAAAEgAAAAQAAAAcAAAAAAAAAEAAAAAAAAAAAkAAAAQAAAAbWluX3ZhbHVl"
+
"AAAAAAAAAAAAAAAAAAABAAAAAAAAAAAJAAAAEAAAAG1heF92YWx1ZQAAAAAAAAAAAAAAAQAAAAAA"
+
"AAAAAAAAAAAAAAIAAAAAAAAAZXh0cmExAIZleHRyYTIAhgMAAAAAAAAAZmllbGQxAIZmaWVsZDIA"
+
"hmZpZWxkMwCGaGRmczovLy9wYXRoL3RvL3dhcmVob3VzZQAAAAAAAAAEAAAAAAAAAGEAAAAAAACB"
+
"YgAAAAAAAIFjAAAAAAAAgWYAAAAAAACBBAAAAAAAAAAPAAAAAAAAAGQAAAAAAAAAlgAAAAAAAADI"
+ "AAAAAAAAAAEAAAABAQANZGVsZXRpb25fZmlsZQAAAAAAAABkAAAAAAAAABYAAAAAAAAAIQAA"
+)
+
+
class SplitSerializerTest(unittest.TestCase):
def _partition_fields(self):
@@ -139,6 +160,51 @@ class SplitSerializerTest(unittest.TestCase):
_decode_modified_utf8(bytes([0xED, 0xA0, 0xBD, 0xED, 0xB8, 0x80])),
'\U0001F600')
+ def test_deserialize_v9_data_and_indexed_splits(self):
+ fields = [DataField(0, 's', AtomicType('STRING'))]
+ for indexed in (False, True):
+ with self.subTest(indexed=indexed):
+ if indexed:
+ data = (_GOLDEN_INDEXED_SPLIT_V1[:28] +
_GOLDEN_DATA_SPLIT_V9
+ + struct.pack('>iqqBif', 1, 13, 14, 1, 1, 0.75))
+ else:
+ data = _GOLDEN_DATA_SPLIT_V1[:16] + _GOLDEN_DATA_SPLIT_V9
+ split = deserialize_split_v1(data, fields, fields)
+ self.assertEqual(split.snapshot_id, 18)
+ self.assertEqual(list(split.partition.values), ['aaaaa'])
+ self.assertEqual(split.bucket, 20)
+ self.assertFalse(split.raw_convertible)
+ self.assertEqual(len(split.files), 1)
+ file = split.files[0]
+ self.assertEqual(file.file_name, 'my_file')
+ self.assertEqual(file.file_path, 'hdfs:///path/to/warehouse')
+ self.assertEqual((file.file_size, file.row_count), (1024 *
1024, 1024))
+ self.assertEqual(list(file.min_key.values), ['min_key'])
+ self.assertEqual(list(file.max_key.values), ['max_key'])
+ self.assertEqual((file.min_sequence_number,
file.max_sequence_number), (15, 200))
+ self.assertEqual((file.schema_id, file.level), (5, 3))
+ self.assertEqual(file.extra_files, ['extra1', 'extra2'])
+ self.assertEqual(file.embedded_index, bytes([1, 2, 4]))
+ self.assertEqual(file.value_stats_cols, ['field1', 'field2',
'field3'])
+ self.assertEqual(file.first_row_id, 12)
+ self.assertEqual(file.write_cols, ['a', 'b', 'c', 'f'])
+ self.assertEqual(len(split.data_deletion_files), 1)
+ dv = split.data_deletion_files[0]
+ self.assertEqual(
+ (dv.dv_index_path, dv.offset, dv.length, dv.cardinality),
+ ('deletion_file', 100, 22, 33))
+ if indexed:
+ self.assertEqual([(r.from_, r.to) for r in
split.row_ranges()], [(13, 14)])
+ self.assertEqual(split.scores(), [0.75])
+
+ def test_unsupported_data_split_version_raises(self):
+ for version in (7, 10):
+ with self.subTest(version=version):
+ data = (_GOLDEN_DATA_SPLIT_V1[:24] + struct.pack('>i', version)
+ + _GOLDEN_DATA_SPLIT_V1[28:])
+ with self.assertRaisesRegex(ValueError, 'unsupported DataSplit
version'):
+ deserialize_split_v1(data, self._partition_fields())
+
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'])
