This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 680fa0e5607a582045c76d333686afac44d7588a 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'])
