diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 54d541a2bcc4..b14c3d5424dc 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 e5354c76968b..47f491a06f17 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 86d77a6782b7..8baf8ab0d41e 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 3bde20330027..a6b1d2b8c387 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 @@ 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) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index 3d9bc985ee8a..60042cbf0c23 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/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: diff --git a/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py b/paimon-python/pypaimon/tests/s3_delete_compatibility_test.py index f5dd6d08ec97..1f7b59e1aad7 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 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"}, 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 000000000000..8e24201cc006 --- /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"), +] + + +@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 diff --git a/paimon-python/pypaimon/tests/table_update_test.py b/paimon-python/pypaimon/tests/table_update_test.py index ddbc4d9a8279..4f832bdb82c4 100644 --- a/paimon-python/pypaimon/tests/table_update_test.py +++ b/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() diff --git a/paimon-python/pyproject.toml b/paimon-python/pyproject.toml index c8761d58561f..ded227052f37 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",