[python] Support newer PyArrow versions

.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

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

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"

paimon-python/pypaimon/execution/python/write/table_update.py

@@ -253,6 +253,8 @@ def _assignment_to_array(
                 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 @@ def _assignment_to_array(
                 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)

paimon-python/pypaimon/filesystem/pyarrow_file_io.py

@@ -557,6 +557,7 @@ def delete(self, path: str, recursive: bool = False) -> bool:
                     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 @@ def _delete_s3_compatible_directory(self, path_str: str) -> bool:
         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 @@ def _get_s3_delete_client(self):
         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:

paimon-python/pypaimon/tests/s3_delete_compatibility_test.py

@@ -24,6 +24,7 @@
 import hashlib
 import os
 import socketserver
+import sys
 import threading
 import unittest
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -299,7 +300,9 @@ def test_pyarrow_22_recursive_delete_batches_objects(self):
         ], [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 @@ def list_page(**kwargs):
         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 @@ def test_recursive_delete_treats_schema_as_regular_key(self):
             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 @@ def test_recursive_delete_treats_schema_as_regular_key(self):
             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 @@ def test_pickle_discards_cached_delete_client(self):
             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 @@ def test_delete_uses_cross_bucket_list_status_path(self):
                     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 @@ def test_delete_uses_cross_bucket_list_status_path(self):
                 ["/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 @@ def test_recursive_delete_control_character_keys(self):
                           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 @@ def test_recursive_delete_retries_partial_batch_failure(self):
         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 @@ def test_recursive_delete_retries_partial_batch_failure(self):
                     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 @@ def test_recursive_delete_reports_late_objects_during_races(self):
         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 @@ def test_recursive_delete_reports_late_objects_during_races(self):
                 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"},

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"),
+]
+
+
+@pytest.mark.parametrize("wrap_type,wrap_value", WRAPPERS)
+@pytest.mark.parametrize("chunked", [False, True])
+@pytest.mark.parametrize("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

paimon-python/pypaimon/tests/table_update_test.py

@@ -555,6 +555,27 @@ def test_update_by_predicate_accepts_array_chunked_and_scalar_values(self):
             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()