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 33a8cd34d3 [python] Support newer PyArrow versions (#9821)
33a8cd34d3 is described below
commit 33a8cd34d3fc23c994cbbe0b1644f61035ee01e6
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Oct 21 15:44:32 2026 +0800
[python] Support newer PyArrow versions (#9821)
---
.github/workflows/ci-python.yml | 11 ++--
docs/docs/pypaimon/catalogs.mdx | 2 +-
paimon-python/dev/lint-python.sh | 2 +-
.../execution/python/write/table_update.py | 29 +++++++++
.../pypaimon/filesystem/pyarrow_file_io.py | 35 +++++++++--
.../pypaimon/tests/s3_delete_compatibility_test.py | 39 ++++++++----
.../tests/table_update_arrow_compatibility_test.py | 73 ++++++++++++++++++++++
paimon-python/pypaimon/tests/table_update_test.py | 21 +++++++
paimon-python/pyproject.toml | 9 ++-
9 files changed, 195 insertions(+), 26 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 54d541a2bc..b14c3d5424 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -165,13 +165,13 @@ jobs:
./paimon-python/dev/lint-python.sh -e pytest_torch
s3-delete-compatibility:
- name: S3 deletion / PyArrow ${{ matrix.pyarrow }}
+ name: Compatibility / PyArrow ${{ matrix.pyarrow }}
runs-on: ubuntu-latest
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
- pyarrow: ['22.0.0', '23.0.0']
+ pyarrow: ['16.1.0', '20.0.0', '21.0.0', '22.0.0']
steps:
- uses: actions/checkout@v6
- uses: actions/setup-python@v5
@@ -181,14 +181,15 @@ jobs:
run: |
python -m pip install --upgrade pip
python -m pip install -e ./paimon-python --group
'./paimon-python/pyproject.toml:test'
- # Exercise the compatibility path before widening the package's
Arrow range.
+ # Exercise each supported PyArrow version.
python -m pip install --no-deps 'pyarrow==${{ matrix.pyarrow }}'
- - name: Test S3 compatibility
+ - name: Test Arrow compatibility
working-directory: paimon-python
run: |
python -m pytest -q pypaimon/tests/s3_delete_compatibility_test.py \
pypaimon/tests/s3_client_compatibility_test.py
pypaimon/tests/oss_legacy_mode_test.py \
- pypaimon/tests/file_io_test.py
pypaimon/tests/filesystem_catalog_test.py
+ pypaimon/tests/file_io_test.py
pypaimon/tests/filesystem_catalog_test.py \
+ pypaimon/tests/table_update_arrow_compatibility_test.py
rust-plan:
name: Native CI
diff --git a/docs/docs/pypaimon/catalogs.mdx b/docs/docs/pypaimon/catalogs.mdx
index e5354c7696..47f491a06f 100644
--- a/docs/docs/pypaimon/catalogs.mdx
+++ b/docs/docs/pypaimon/catalogs.mdx
@@ -48,7 +48,7 @@ catalog_options = {
catalog = CatalogFactory.create(catalog_options)
```
-For an S3 warehouse, pass S3 authentication options with the filesystem
catalog options:
+For an S3 warehouse, install `pypaimon[s3]` and pass S3 authentication options
with the filesystem catalog options:
```python
from pypaimon import CatalogFactory
diff --git a/paimon-python/dev/lint-python.sh b/paimon-python/dev/lint-python.sh
index 86d77a6782..8baf8ab0d4 100755
--- a/paimon-python/dev/lint-python.sh
+++ b/paimon-python/dev/lint-python.sh
@@ -198,7 +198,7 @@ function pytest_check() {
# 3.6/3.7 run a curated core subset (their dep ceiling rules out the
# vector/index/multimodal/blob suites); 3.10+ run the full suite.
if [ "$PYTHON_VERSION" = "3.6" ] || [ "$PYTHON_VERSION" = "3.7" ]; then
- TEST_DIR="pypaimon/tests/py36 pypaimon/tests/file_io_test.py
pypaimon/tests/arrow_schema_test.py"
+ TEST_DIR="pypaimon/tests/py36 pypaimon/tests/file_io_test.py
pypaimon/tests/arrow_schema_test.py
pypaimon/tests/table_update_arrow_compatibility_test.py"
echo "Running core test subset for Python $PYTHON_VERSION: $TEST_DIR"
else
TEST_DIR="pypaimon/tests pypaimon/acceptance
--ignore=pypaimon/tests/py36 --ignore=pypaimon/tests/e2e
--ignore=pypaimon/tests/torch_read_test.py"
diff --git a/paimon-python/pypaimon/execution/python/write/table_update.py
b/paimon-python/pypaimon/execution/python/write/table_update.py
index 3bde203300..a6b1d2b8c3 100644
--- a/paimon-python/pypaimon/execution/python/write/table_update.py
+++ b/paimon-python/pypaimon/execution/python/write/table_update.py
@@ -253,6 +253,8 @@ class PythonTableUpdateMixin:
f"{len(array)} != {row_count}."
)
if array.type != data_type:
+ PythonTableUpdateMixin._validate_assignment_struct_fields(
+ array.type, data_type)
if isinstance(array, pa.ChunkedArray):
array = pa.chunked_array(
[chunk.cast(data_type) for chunk in array.chunks],
@@ -262,6 +264,33 @@ class PythonTableUpdateMixin:
array = array.cast(data_type)
return array
+ @staticmethod
+ def _validate_assignment_struct_fields(source, target):
+ if pa.types.is_struct(source) and pa.types.is_struct(target):
+ source_fields = {field.name: field for field in source}
+ target_names = {field.name for field in target}
+ shared_names = source_fields.keys() & target_names
+ if ([field.name for field in source if field.name in shared_names]
+ != [field.name for field in target if field.name in
shared_names]):
+ raise ValueError("Assignment struct fields must match in
order.")
+ for target_field in target:
+ source_field = source_fields.get(target_field.name)
+ if source_field is not None:
+ PythonTableUpdateMixin._validate_assignment_struct_fields(
+ source_field.type, target_field.type)
+ elif pa.types.is_map(source) and pa.types.is_map(target):
+ PythonTableUpdateMixin._validate_assignment_struct_fields(
+ source.key_type, target.key_type)
+ PythonTableUpdateMixin._validate_assignment_struct_fields(
+ source.item_type, target.item_type)
+ else:
+ list_types = (pa.types.is_list, pa.types.is_large_list,
+ pa.types.is_fixed_size_list)
+ if (any(check(source) for check in list_types)
+ and any(check(target) for check in list_types)):
+ PythonTableUpdateMixin._validate_assignment_struct_fields(
+ source.value_type, target.value_type)
+
def _delete_by_partition_filter(
self, partition_filter: Predicate) -> List[CommitMessage]:
_authorize_update(self.table)
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 3d9bc985ee..60042cbf0c 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -557,6 +557,7 @@ class PyArrowFileIO(FileIO):
raise OSError(f"Directory {path} is not empty")
bucket, key = self._split_s3_path(path_str)
if key:
+
self._ensure_s3_parent_exists(self._get_s3_delete_client(), bucket, key)
self._get_s3_delete_client().delete_object(
Bucket=bucket, Key=key.rstrip("/") + "/")
return True
@@ -581,18 +582,35 @@ class PyArrowFileIO(FileIO):
if prefix:
prefix += "/"
deadline = time.monotonic() + _S3_DELETE_TIMEOUT_SECONDS
- # Finish listing before deleting: mutating a paginated listing may
skip keys.
+ root_schema = prefix + "schema/schema-0"
+ has_root_schema = False
+ # Finish listing before deleting; retain only the root schema-0.
with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as listed:
for name in self._list_s3_keys(client, bucket, prefix, deadline,
path_str):
- listed.write(json.dumps(name) + "\n")
+ if name == root_schema:
+ has_root_schema = True
+ else:
+ listed.write(json.dumps(name) + "\n")
listed.seek(0)
self._delete_s3_objects(
client, bucket, (json.loads(line) for line in listed),
deadline, path_str)
- if any(self._list_s3_keys(client, bucket, prefix, deadline, path_str)):
- raise OSError(f"S3 directory {path_str} changed during deletion")
+ for name in self._list_s3_keys(client, bucket, prefix, deadline,
path_str):
+ if not has_root_schema or name != root_schema:
+ raise OSError(f"S3 directory {path_str} changed during
deletion")
+ self._check_s3_delete_deadline(deadline, path_str)
+ self._ensure_s3_parent_exists(client, bucket, key)
+ if has_root_schema:
+ self._delete_s3_objects(client, bucket, [root_schema], deadline,
path_str)
return True
+ @staticmethod
+ def _ensure_s3_parent_exists(client, bucket: str, key: str):
+ parent, _, _ = key.rstrip("/").rpartition("/")
+ if parent:
+ client.put_object(Bucket=bucket, Key=parent + "/", Body=b"",
+ ContentType="application/x-directory")
+
def _list_s3_keys(self, client, bucket: str, prefix: str,
deadline: float, path_str: str):
token = None
@@ -670,7 +688,14 @@ class PyArrowFileIO(FileIO):
if self._s3_delete_client is not None:
return self._s3_delete_client
- import boto3
+ try:
+ import boto3
+ except ModuleNotFoundError as error:
+ if error.name != "boto3":
+ raise
+ raise ImportError(
+ "S3/OSS batch deletion requires pypaimon[s3] or pypaimon[oss]."
+ ) from error
from botocore.config import Config
if self._is_oss:
diff --git a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
index f5dd6d08ec..1f7b59e1aa 100644
--- a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
+++ b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py
@@ -24,6 +24,7 @@ import base64
import hashlib
import os
import socketserver
+import sys
import threading
import unittest
from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -299,7 +300,9 @@ class OssDeleteCompatibilityTest(unittest.TestCase):
], [item["Key"] for item in
client.delete_objects.call_args[1]["Delete"]["Objects"]])
client.delete_object.assert_not_called()
- client.put_object.assert_not_called()
+ client.put_object.assert_called_once_with(
+ Bucket="test-bucket", Key="db-uuid.db/", Body=b"",
+ ContentType="application/x-directory")
file_io.filesystem.delete_file.assert_not_called()
file_io.filesystem.delete_dir_contents.assert_not_called()
file_io.filesystem.delete_dir.assert_not_called()
@@ -332,7 +335,7 @@ class OssDeleteCompatibilityTest(unittest.TestCase):
self.assertEqual("next", client.list_objects_v2.call_args_list[1][1][
"ContinuationToken"])
- def test_recursive_delete_treats_schema_as_regular_key(self):
+ def test_recursive_delete_retains_schema_zero_until_validation(self):
file_io = self._new_file_io(legacy=False)
file_io._pyarrow_gte_22 = True
directory = file_io.to_filesystem_path(TABLE_PATH)
@@ -344,7 +347,7 @@ class OssDeleteCompatibilityTest(unittest.TestCase):
prefix + "snapshot/snapshot-1",
prefix + "schema/schema-1",
prefix + "data/file.parquet",
- ], [])
+ ], [prefix + "schema/schema-0"])
self.assertTrue(file_io.delete(TABLE_PATH, recursive=True))
@@ -356,6 +359,10 @@ class OssDeleteCompatibilityTest(unittest.TestCase):
prefix + "data/file.parquet",
], calls)
+ self.assertEqual(
+ [{"Key": prefix + "schema/schema-0"}],
+
file_io._s3_delete_client.delete_objects.call_args[1]["Delete"]["Objects"])
+
def test_recursive_delete_times_out_when_directory_keeps_changing(self):
file_io = self._new_file_io(legacy=False)
file_io._pyarrow_gte_22 = True
@@ -572,6 +579,13 @@ class CustomS3EndpointTest(unittest.TestCase):
restored.__setstate__(state)
self.assertIsNone(restored._s3_delete_client)
+ def test_missing_boto3_reports_optional_extra(self):
+ file_io = self._new_file_io()
+ file_io._s3_delete_client = None
+ with mock.patch.dict(sys.modules, {"boto3": None}):
+ with self.assertRaisesRegex(ImportError, r"pypaimon\[s3\]"):
+ file_io._get_s3_delete_client()
+
def test_jindo_and_native_s3_do_not_use_delete_fallback(self):
file_io = self._new_file_io()
file_io._pyarrow_gte_22 = True
@@ -709,7 +723,7 @@ class CustomS3EndpointTest(unittest.TestCase):
for recursive in (False, True):
with self.subTest(path=path, recursive=recursive):
server.bucket_objects["target-bucket"] = {
- "parent/", "parent/child/"}
+ "parent/child/"}
if recursive:
server.bucket_objects["target-bucket"].add(
"parent/child/delete.parquet")
@@ -737,7 +751,7 @@ class CustomS3EndpointTest(unittest.TestCase):
["/target-bucket"] * 4,
[urlsplit(path).path for method, path in server.requests
if method == "POST"])
- self.assertEqual([], server.put_headers)
+ self.assertEqual(8, len(server.put_headers))
finally:
server.shutdown()
server.server_close()
@@ -793,7 +807,7 @@ class CustomS3EndpointTest(unittest.TestCase):
if method == "DELETE"]
self.assertIn("/test-bucket/" + carriage_return, individual)
self.assertIn("/test-bucket/" + control, individual)
- self.assertEqual(1, sum(
+ self.assertEqual(2, sum(
method == "POST" for method, _ in server.requests))
finally:
server.shutdown()
@@ -810,7 +824,7 @@ class CustomS3EndpointTest(unittest.TestCase):
server.requests = []
server.prefix = "parent/table/"
deleted = server.prefix + "data/deleted.parquet"
- failed = server.prefix + "data/failed.parquet"
+ failed = server.prefix + "branch/branch-dev/schema/schema-0"
schema_zero = server.prefix + "schema/schema-0"
server.objects = {"parent/", server.prefix, deleted, failed,
schema_zero}
server.fail_delete_key_once = failed
@@ -842,13 +856,14 @@ class CustomS3EndpointTest(unittest.TestCase):
file_io.delete(
"s3://test-bucket/parent/table", recursive=True)
self.assertIn(failed, str(error.exception))
- self.assertEqual({"parent/", failed}, server.objects)
+ self.assertEqual({"parent/", failed,
"parent/table/schema/schema-0"},
+ server.objects)
self.assertTrue(file_io.delete(
"s3://test-bucket/parent/table", recursive=True))
file_io._s3_delete_client.close()
self.assertEqual({"parent/"}, server.objects)
- self.assertEqual(2, sum(
+ self.assertEqual(3, sum(
method == "POST" for method, _ in server.requests))
finally:
server.shutdown()
@@ -866,7 +881,8 @@ class CustomS3EndpointTest(unittest.TestCase):
server.prefix = "ta/ble/"
decoy = "ta/ble-other/keep.parquet"
server.objects = {
- "ta/", server.prefix, server.prefix + "first.parquet", decoy}
+ "ta/", server.prefix, server.prefix + "first.parquet",
+ server.prefix + "schema/schema-0", decoy}
server.inject_late_after_delete = server.prefix + "first.parquet"
server.late_object_added = False
server.missing_object_removed = False
@@ -898,7 +914,8 @@ class CustomS3EndpointTest(unittest.TestCase):
file_io._s3_delete_client.close()
self.assertTrue(server.missing_object_removed)
- self.assertEqual({decoy, "ta/", server.prefix + "late.parquet"},
+ self.assertEqual({decoy, "ta/", server.prefix + "late.parquet",
+ server.prefix + "schema/schema-0"},
server.objects)
self.assertEqual(
{"GET", "POST"},
diff --git
a/paimon-python/pypaimon/tests/table_update_arrow_compatibility_test.py
b/paimon-python/pypaimon/tests/table_update_arrow_compatibility_test.py
new file mode 100644
index 0000000000..8e24201cc0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/table_update_arrow_compatibility_test.py
@@ -0,0 +1,73 @@
+# 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 pyarrow as pa
+import pytest
+
+from pypaimon.write.table_update import BatchTableUpdate
+
+
+WRAPPERS = [
+ pytest.param(lambda t: t, lambda v: v, id="struct"),
+ pytest.param(lambda t: pa.struct([('outer', t)]),
+ lambda v: {'outer': v}, id="nested-struct"),
+ pytest.param(pa.list_, lambda v: [v], id="list"),
+ pytest.param(pa.large_list, lambda v: [v], id="large-list"),
+ pytest.param(lambda t: pa.list_(t, 1), lambda v: [v], id="fixed-list"),
+ pytest.param(lambda t: pa.map_(pa.string(), t),
+ lambda v: [('key', v)], id="map-value"),
+ pytest.param(lambda t: pa.list_(pa.map_(pa.string(), t)),
+ lambda v: [[('key', v)]], id="list-map-value"),
+]
+
+
[email protected]("wrap_type,wrap_value", WRAPPERS)
[email protected]("chunked", [False, True])
[email protected]("projection", [
+ "same", "subset-first", "subset-last", "add-field", "reordered",
+])
+def test_predicate_assignment_struct_cast(wrap_type, wrap_value, chunked,
projection):
+ source = wrap_type(pa.struct([('a', pa.int32()), ('b', pa.int32())]))
+ fields = [('a', pa.int64()), ('b', pa.int64())]
+ if projection == "subset-first":
+ fields = fields[:1]
+ elif projection == "subset-last":
+ fields = fields[1:]
+ elif projection == "add-field":
+ fields.append(('c', pa.int64()))
+ elif projection == "reordered":
+ fields.reverse()
+ target = wrap_type(pa.struct(fields))
+ value = wrap_value({'a': 1, 'b': 2})
+ values = pa.array([value, None], type=source)
+ if chunked:
+ values = pa.chunked_array([values.slice(0, 1), values.slice(1)])
+
+ if projection == "reordered":
+ with pytest.raises(ValueError, match="struct fields must match in
order"):
+ BatchTableUpdate._assignment_to_array(values, target, 2)
+ else:
+ try:
+ expected = values.cast(target).to_pylist()
+ except pa.ArrowNotImplementedError:
+ # Older Arrow versions do not implement STRUCT casts.
+ with pytest.raises(pa.ArrowNotImplementedError):
+ BatchTableUpdate._assignment_to_array(values, target, 2)
+ return
+ result = BatchTableUpdate._assignment_to_array(values, target, 2)
+ assert result.type == target
+ assert result.to_pylist() == expected
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index ddbc4d9a82..4f832bdb82 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -555,6 +555,27 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
rows,
)
+ def test_update_by_predicate_accepts_struct_subset_assignment(self):
+ schema = pa.schema([
+ ('id', pa.int32()),
+ ('payload', pa.struct([('a', pa.int64())])),
+ ])
+ source_type = pa.struct([('a', pa.int32()), ('b', pa.int32())])
+ for chunked in (False, True):
+ with self.subTest(chunked=chunked):
+ table = self._create_table(pa_schema=schema)
+ self._write_arrow(table, pa.Table.from_pylist(
+ [{'id': 1, 'payload': {'a': 0}}], schema=schema))
+ values = pa.array([{'a': 7, 'b': 99}], type=source_type)
+ if chunked:
+ values = pa.chunked_array([values])
+ predicate =
table.new_read_builder().new_predicate_builder().equal('id', 1)
+
+ self._do_update_by_predicate(table, predicate, {'payload':
values})
+
+ self.assertEqual(
+ [{'a': 7}], self._read_all(table)['payload'].to_pylist())
+
def test_update_by_predicate_accepts_callable_assignments(self):
table = self._create_seeded_table()
pb = table.new_read_builder().new_predicate_builder()
diff --git a/paimon-python/pyproject.toml b/paimon-python/pyproject.toml
index c8761d5856..ded227052f 100644
--- a/paimon-python/pyproject.toml
+++ b/paimon-python/pyproject.toml
@@ -63,7 +63,7 @@ dependencies = [
"polars>=1.32,<2; python_version >= '3.9'",
"pyarrow>=6,<7; python_version < '3.7'",
"pyarrow>=7,<13; python_version >= '3.7' and python_version < '3.8'",
- "pyarrow>=16,!=19.0.0,<20; python_version >= '3.8'",
+ "pyarrow>=16,!=19.0.0,<23; python_version >= '3.8'",
"pyroaring<=0.3.3; python_version < '3.7'",
"pyroaring<=0.4.5; python_version == '3.7'",
"pyroaring>=1.0.0,<=1.0.3; python_version == '3.8'",
@@ -97,10 +97,9 @@ hdf5 = [
rosbag = [
"rosbags>=0.11.5,<0.12; python_version >= '3.10'",
]
-# datasets 4.1+ may require PyArrow 21+, outside the supported core range.
# Pandas 2.2.2+ supports NumPy 2 selected by the media dependencies.
lerobot = [
- "datasets>=4,<4.1; python_version >= '3.10'",
+ "datasets>=4,<5; python_version >= '3.10'",
"pandas>=2.2.2,<3; python_version >= '3.10'",
"torch>=2.3; python_version >= '3.10'",
"lerobot>=0.4.4,<0.5; python_version >= '3.10'",
@@ -121,10 +120,14 @@ daft = [
"daft>=0.7.6; python_version >= '3.10'",
]
oss = [
+ "boto3>=1.34,<2; python_version >= '3.10'",
"oss2>=2.18,<3",
"ossfs>=2021.8; python_version < '3.8'",
"ossfs>=2023; python_version >= '3.8'",
]
+s3 = [
+ "boto3>=1.34,<2; python_version >= '3.10'",
+]
jindo = [
"oss2>=2.18,<3",
"pyjindosdk>=6.10.4",
