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 021ca8c10c [Python] Complete native mutation routing and typed
partition commits (#10653)
021ca8c10c is described below
commit 021ca8c10cf35444e5a79dafdb0a729a812f0b81
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 18:05:39 2026 +0800
[Python] Complete native mutation routing and typed partition commits
(#10653)
---
docs/docs/pypaimon/writing.md | 9 +
.../tests/native_commit_partition_spec_test.py | 266 +++++++++++++++++++++
.../tests/native_manifest_compression_test.py | 170 +++++++++++++
.../native_partition_delete_changelog_test.py | 104 ++++++++
.../tests/native_upsert_encoded_keys_test.py | 127 +++++++++-
.../pypaimon/tests/partition_predicate_test.py | 2 +-
paimon-python/pypaimon/write/file_store_commit.py | 66 ++---
paimon-python/pypaimon/write/native_update.py | 66 ++++-
paimon-python/pypaimon/write/partition_spec.py | 226 +++++++++++++++++
9 files changed, 998 insertions(+), 38 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index 33144b6dd4..e84209ef6c 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -82,6 +82,11 @@ prepare-and-commit lifecycle. Native commits also support
overwrite,
Overwrite and truncate record an OVERWRITE snapshot even when no files match,
and remove index metadata for the replaced partitions. Physical files remain
available to historical snapshots. Changelog-only commits also publish a
snapshot.
+For Parquet tables, both paths support `manifest.compression=null`, `deflate`,
+`snappy`, `zstd`, `bzip2` and `xz`, including manifests retained from earlier
+snapshots. Overwrite and truncate partition specs convert textual values to the
+declared partition type. `None` and the configured default partition name
select
+NULL partitions; malformed values are rejected before publication.
Standard REST tables with `query-auth.enabled=true` can write and commit
without requesting SELECT permission. The REST server still authorizes snapshot
@@ -95,6 +100,10 @@ grant. MERGE table sources obtain their own grants
independently from the target
including historical snapshots and branches.
Query-authorized cross-partition index bootstrap remains unsupported.
+Whole-partition predicate deletes can use Native on primary-key tables
configured
+with input changelogs. Native upserts also accept fully NULL source key columns
+with a different physical type, including selected NULL dictionary values.
+
`write_ray()` uses the shared Python coordinator and lets workers select Native
writers. For postpone fixed-bucket writes, overwrite plans use the new input
unless an explicit bucket plan was supplied. A complete primary-key
diff --git a/paimon-python/pypaimon/tests/native_commit_partition_spec_test.py
b/paimon-python/pypaimon/tests/native_commit_partition_spec_test.py
new file mode 100644
index 0000000000..ab3bb96cdb
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_commit_partition_spec_test.py
@@ -0,0 +1,266 @@
+# 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.
+
+"""Public overwrite/truncate specs use Java's declared partition types."""
+
+from datetime import date, datetime, time, timedelta, timezone
+from decimal import Decimal
+from pathlib import Path
+from unittest.mock import patch
+from urllib.parse import unquote, urlparse
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.partition_spec import _partition_spec_value
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+rest_catalog = native_plan_rest_test.rest_catalog
+native = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_ENGINES = [False, pytest.param(True, marks=native)]
+_CASES = [
+ ('integer', pa.int32(), [1, 2], '1'),
+ ('typed-integer', pa.int32(), [1, 2], 1),
+ ('unicode-integer', pa.int32(), [12, 13], '１２'),
+ ('boolean', pa.bool_(), [True, False], 'TRUE'),
+ ('string', pa.string(), ['a/b = 中', 'other'], 'a/b = 中'),
+ ('decimal', pa.decimal128(6, 2), [Decimal('12.35'), Decimal('13.00')],
'12.345'),
+ ('unicode-decimal', pa.decimal128(6, 2), [Decimal('12.35'),
Decimal('13.00')], '１２.３４５'),
+ ('decimal-overflow', pa.decimal128(3, 0), [None, Decimal('12')], '1000'),
+ ('date', pa.date32(), [date(2024, 1, 2), date(2024, 1, 3)], '2024-01-02'),
+ ('date-partial', pa.date32(), [date(2024, 1, 1), date(2024, 1, 3)],
'2024-1'),
+ ('date-epoch', pa.date32(), [date(1970, 1, 2), date(1970, 1, 3)], '1'),
+ ('time', pa.time32('ms'), [time(12, 34, 56, 789000), time(12, 35)],
'12:34:56.789'),
+ ('time-rounding', pa.time32('ms'), [time(12, 34, 56, 790000), time(12,
35)], '12:34:56.7895'),
+ ('time-fraction', pa.time32('ms'), [time(12, 34, 56, 12000), time(12,
35)], '12:34:56.\u00a0123'),
+ ('time-fraction-utf16', pa.time32('ms'), [time(12, 34, 56, 1000), time(12,
34, 56, 12000)],
+ '12:34:56.😀123'),
+ ('time-offset', pa.time32('ms'), [time(13, 34, 56), time(13, 35)],
'12:34:56+01:00'),
+ ('time-epoch', pa.time32('ms'), [time(0, 0, 0, 1000), time(0, 0, 0,
2000)], '1'),
+ ('timestamp', pa.timestamp('us'), [datetime(2024, 1, 2, 12, 34),
datetime(2024, 1, 3)],
+ '2024-01-02 12:34:00'),
+ ('timestamp-iso', pa.timestamp('us'), [datetime(2024, 1, 2, 12, 34),
datetime(2024, 1, 3)],
+ '2024-01-02T12:34'),
+ ('timestamp-repeated', pa.timestamp('us'), [datetime(2024, 1, 2, 12, 34),
datetime(2024, 1, 3)],
+ '2024-0101-0202 1212:3434:00T12:34'),
+ ('timestamp-defaults', pa.timestamp('us'), [datetime(2024, 1, 1),
datetime(2024, 1, 3)], '2024--'),
+ ('timestamp-nanos', pa.timestamp('ns'), [datetime(2024, 1, 2, 12, 34, 0,
123456), datetime(2024, 1, 3)],
+ '2024-01-02 12:34:00.123456000'),
+ ('timestamp-overflow', pa.timestamp('s'), [datetime(1969, 12, 31, 23, 59,
59), datetime(1970, 1, 1)],
+ '9223372036854775807'),
+ ('timestamp-ltz', pa.timestamp('us', tz='UTC'),
+ [datetime(2024, 1, 2, 12, 34, tzinfo=timezone.utc), datetime(2024, 1, 3,
tzinfo=timezone.utc)],
+ '2024-01-02 12:34:00'),
+ ('float', pa.float32(), [1.2, 2.0], '1.2'),
+ ('float-hex', pa.float32(), [4.0, 2.0], '0x1.p2'),
+ ('float-hex-overflow-positive', pa.float32(), [float('inf'), 2.0],
'0x1p1024'),
+ ('float-hex-overflow-negative', pa.float32(), [float('-inf'), 2.0],
'-0x1p1024'),
+ ('double-hex-overflow-positive', pa.float64(), [float('inf'), 2.0],
'0x1p1024'),
+ ('double-hex-overflow-negative', pa.float64(), [float('-inf'), 2.0],
'-0x1p1024'),
+ ('float-trim', pa.float32(), [1.2, 2.0], '\x001.2\x1f'),
+ ('null', pa.int32(), [None, 2], None),
+ ('default-name', pa.int32(), [None, 2], '__DEFAULT_PARTITION__'),
+]
+
+
+def _table(catalog, data_type, values, use_native):
+ schema = pa.schema([('id', pa.int64()), ('pt', data_type)])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['pt'], options={
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false',
+ 'dynamic-partition-overwrite': 'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [1, 2], 'pt': values},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table.copy({'commit.native.enabled': str(use_native).lower()}),
schema
+
+
+def _ids(table):
+ builder = table.new_read_builder()
+ return sorted(row['id'] for row in
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist())
+
+
+def _files(table):
+ return set(Path(unquote(urlparse(table.table_path).path)).rglob('*'))
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['truncate', 'overwrite'])
[email protected]('name,data_type,values,text', _CASES, ids=[case[0]
for case in _CASES])
+def test_partition_specs_parse_declared_types(rest_catalog, use_native,
operation, name, data_type, values, text):
+ catalog, _ = rest_catalog
+ if name == 'timestamp-ltz':
+ # Java treats this text as process-local wall time, not as UTC.
+ text = values[0].astimezone().replace(tzinfo=None).strftime('%Y-%m-%d
%H:%M:%S')
+ table, schema = _table(catalog, data_type, values, use_native)
+ builder = table.new_batch_write_builder()
+ writer = None
+ if operation == 'overwrite':
+ builder.overwrite({'pt': text})
+ writer = builder.new_write()
+ writer.write_arrow(pa.table({'id': [3], 'pt': values[:1]},
schema=schema))
+ commit = builder.new_commit()
+ try:
+ if use_native:
+ method = 'drop_partitions' if operation == 'truncate' else
'overwrite'
+ guard = patch.object(commit.file_store_commit, method,
side_effect=AssertionError('Python fallback'))
+ else:
+ guard = patch.object(commit, '_prepare_native_commit',
return_value=None)
+ with guard:
+ if operation == 'truncate':
+ commit.truncate_partitions([{'pt': text}])
+ else:
+ commit.commit(writer.prepare_commit())
+ assert _ids(table) == ([2] if operation == 'truncate' else [2, 3])
+ assert table.snapshot_manager().get_latest_snapshot().id == 2
+ finally:
+ if writer is not None:
+ writer.close()
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['truncate', 'overwrite'])
[email protected]('spec', [{'missing': '1'}, {'pt': 'bad'}, {'pt':
'1.5'}, {'pt': '2147483648'},
+ {'pt': ' 1'}, {'pt': '1_0'}, {'pt': '𝟙'}])
+def test_invalid_specs_do_not_publish_or_delete(rest_catalog, use_native,
operation, spec):
+ catalog, server = rest_catalog
+ table, _ = _table(catalog, pa.int32(), [1, 2], use_native)
+ builder = table.new_batch_write_builder()
+ if operation == 'overwrite':
+ builder.overwrite(spec)
+ commit = builder.new_commit()
+ before = _files(table)
+ try:
+ with patch.object(server, '_table_commit_handle',
wraps=server._table_commit_handle) as publish:
+ with pytest.raises((ValueError, RuntimeError)):
+ if operation == 'truncate':
+ commit.truncate_partitions([spec])
+ else:
+ commit.commit([])
+ publish.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ assert _ids(table) == [1, 2]
+ assert _files(table) == before
+ finally:
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('data_type,values,text', [
+ (pa.bool_(), [True, False], 'on'),
+ (pa.float32(), [1.0, 2.0], '1.23456789'),
+ (pa.decimal128(6, 2), [Decimal('1'), Decimal('2')], 'not a decimal'),
+ (pa.decimal128(6, 2), [Decimal('1'), Decimal('2')], '𝟙.3'),
+ (pa.date32(), [date(2024, 1, 1), date(2024, 1, 2)], '2024-02-30'),
+ (pa.date32(), [date(2024, 1, 1), date(2024, 1, 2)], '2024-\u00a01-01'),
+ (pa.time32('ms'), [time(12, 34), time(12, 35)], '12:34+02:30'),
+ (pa.time32('ms'), [time(12, 34), time(12, 35)], '\u00a012:34:00'),
+ (pa.timestamp('us'), [datetime(2024, 1, 1), datetime(2024, 1, 2)],
'2024-01-01 12:34'),
+ (pa.timestamp('us'), [datetime(2024, 1, 1), datetime(2024, 1, 2)],
'2024-00-01'),
+ (pa.timestamp('us'), [datetime(2024, 1, 1), datetime(2024, 1, 2)],
'2024-01-00'),
+])
+def test_invalid_typed_text_does_not_publish(rest_catalog, use_native,
data_type, values, text):
+ catalog, server = rest_catalog
+ table, _ = _table(catalog, data_type, values, use_native)
+ commit = table.new_batch_write_builder().new_commit()
+ before = _files(table)
+ try:
+ with patch.object(server, '_table_commit_handle',
wraps=server._table_commit_handle) as publish:
+ with pytest.raises((ValueError, RuntimeError)):
+ commit.truncate_partitions([{'pt': text}])
+ publish.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ assert _ids(table) == [1, 2]
+ assert _files(table) == before
+ finally:
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
+def test_overwrite_partition_mismatch_preserves_prepared_files(rest_catalog,
use_native):
+ catalog, server = rest_catalog
+ table, schema = _table(catalog, pa.int32(), [1, 2], use_native)
+ builder = table.new_batch_write_builder().overwrite({'pt': '2'})
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [3], 'pt': [1]}, schema=schema))
+ messages = writer.prepare_commit()
+ before = _files(table)
+ with patch.object(server, '_table_commit_handle',
wraps=server._table_commit_handle) as publish:
+ with pytest.raises((ValueError, RuntimeError)):
+ commit.commit(messages)
+ publish.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ assert _ids(table) == [1, 2]
+ assert _files(table) == before
+ finally:
+ writer.close()
+ commit.close()
+
+
+def test_timestamp_spec_precision_is_independent_of_arrow_storage():
+ assert _partition_spec_value('1970-01-01 00:00:00.123456',
AtomicType('TIMESTAMP(1)')) == (
+ datetime(1970, 1, 1, 0, 0, 0, 100000))
+ value = _partition_spec_value('1970-01-01 00:00:00.123456789',
AtomicType('TIMESTAMP(9)'))
+ assert pa.scalar(value, type=pa.timestamp('ns')).value == 123456789
+
+
[email protected]('sign', [-1, 1])
+def test_numeric_timestamp_overflow_retains_millisecond_remainder(sign):
+ value = _partition_spec_value(str(sign * 18446744073709552),
AtomicType('TIMESTAMP(0)'))
+ assert value == datetime(1970, 1, 1) + timedelta(milliseconds=sign * 384)
+
+
[email protected]('use_native', _ENGINES)
[email protected]('sign', [-1, 1])
+def test_overflow_spec_does_not_truncate_another_partition(rest_catalog,
use_native, sign):
+ catalog, _ = rest_catalog
+ table, _ = _table(catalog, pa.timestamp('s'),
+ [datetime(1970, 1, 1), datetime(1969, 12, 31, 23, 59,
59)], use_native)
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.truncate_partitions([{'pt': str(sign * 18446744073709552)}])
+ assert _ids(table) == [1, 2]
+ assert table.snapshot_manager().get_latest_snapshot().id == 2
+ finally:
+ commit.close()
+
+
[email protected]('data_type,text', [
+ (AtomicType('VARCHAR(2)', False), 'abc'),
+ (AtomicType('VARBINARY(2)', False), '中'),
+ (AtomicType('TINYINT'), '128'),
+ (AtomicType('SMALLINT'), '-32769'),
+ (AtomicType('BIGINT'), '9223372036854775808'),
+])
+def test_partition_spec_type_limits_are_validated(data_type, text):
+ with pytest.raises(ValueError):
+ _partition_spec_value(text, data_type)
diff --git a/paimon-python/pypaimon/tests/native_manifest_compression_test.py
b/paimon-python/pypaimon/tests/native_manifest_compression_test.py
new file mode 100644
index 0000000000..f8bc9ff9ee
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_manifest_compression_test.py
@@ -0,0 +1,170 @@
+# 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.
+
+"""Parquet tables retain compressed manifest history across Pure/Native
commits."""
+
+from contextlib import ExitStack
+from io import BytesIO
+from unittest.mock import Mock, patch
+
+import fastavro
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.tests.native_commit_semantics_test import _publication
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed'))]
+_SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.string()), ('value',
pa.int64())])
+_SEED = [dict(id=1, pt='a', value=10), dict(id=2, pt='b', value=20)]
+
+
+def _prepare(builder, rows, identifier=None):
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ return writer.prepare_commit() if identifier is None else
writer.prepare_commit(identifier)
+ finally:
+ writer.close()
+
+
+def _read(table, native, snapshot_id=None):
+ options = {'scan.native-plan.enabled': str(native).lower(),
'read.native.enabled': str(native).lower()}
+ if snapshot_id is not None:
+ options['scan.snapshot-id'] = str(snapshot_id)
+ builder = table.copy(options).new_read_builder()
+ scan = builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Python plan fallback')))
+ plan = scan.plan()
+ if native:
+ assert all(getattr(split, '_native_split', None) is not None for
split in plan.splits())
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python read
fallback')))
+ return
builder.new_read().to_arrow(plan.splits()).sort_by('id').to_pylist()
+
+
+def _assert_codec(table, snapshot, codec, primary_key):
+ def records(name):
+ with table.file_io.new_input_stream(table.table_path.rstrip('/') +
'/manifest/' + name) as stream:
+ reader = fastavro.reader(BytesIO(stream.read()))
+ assert reader.metadata['avro.codec'] == codec
+ return list(reader)
+
+ lists = ManifestListManager(table)
+ for name in [snapshot.base_manifest_list, snapshot.delta_manifest_list,
snapshot.changelog_manifest_list]:
+ if name is not None:
+ records(name)
+ for manifest in lists.read(name):
+ entries = records(manifest.file_name)
+ assert all(entry['_FILE']['_FILE_NAME'].endswith('.parquet')
for entry in entries)
+ if primary_key and snapshot.total_record_count:
+ assert snapshot.index_manifest
+ assert records(snapshot.index_manifest)
+
+
[email protected]('use_native', _ENGINES)
[email protected]('codec', ['bzip2', 'xz'])
[email protected]('primary_key', [False, True], ids=['append',
'dynamic-pk'])
+def test_compressed_manifest_history_survives_native_publication_and_reads(
+ rest_catalog, use_native, codec, primary_key):
+ catalog, server = rest_catalog
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['pt'], primary_keys=['id', 'pt'] if
primary_key else [], options={
+ 'manifest.compression': codec, 'bucket': '-1',
'dynamic-bucket.target-row-num': '100',
+ 'changelog-producer': 'input' if primary_key else 'none',
+ 'dynamic-partition-overwrite': 'false',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false',
+ 'commit.min-retry-wait': '0 ms', 'commit.max-retry-wait': '0 ms',
+ }), False)
+ table = catalog.get_table('default.t')
+ seed = table.new_batch_write_builder()
+ commit = seed.new_commit()
+ try:
+ commit.commit(_prepare(seed, _SEED))
+ finally:
+ commit.close()
+ first = table.snapshot_manager().get_latest_snapshot()
+ _assert_codec(table, first, codec, primary_key)
+ assert _read(table, use_native) == _SEED
+
+ table = table.copy({'commit.native.enabled': str(use_native).lower()})
+ builder = table.new_stream_write_builder()
+ commit = builder.new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ expected = list(_SEED)
+ try:
+ for identifier, row in [(11, dict(id=3, pt='a', value=30)),
+ (12, dict(id=4, pt='c' if primary_key else
None, value=40))]:
+ messages = _prepare(builder, [row], identifier)
+ with _publication(server, commit, use_native,
lose_response=identifier == 12) as attempts:
+ commit.commit(messages, identifier,
snapshot_properties={'codec': codec})
+ assert attempts == [200]
+ expected.append(row)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_identifier == identifier
+ assert snapshot.properties == {'codec': codec}
+ _assert_codec(table, snapshot, codec, primary_key)
+ assert _read(table, use_native) == expected
+ assert callback.call.call_count == 2
+ finally:
+ commit.close()
+ callback.close.assert_called_once()
+ history = table.snapshot_manager().get_latest_snapshot()
+
+ replacement = dict(id=9, pt='a', value=90)
+ overwrite = table.new_batch_write_builder().overwrite({'pt': 'a'})
+ commit = overwrite.new_commit()
+ try:
+ with _publication(server, commit, use_native):
+ commit.commit(_prepare(overwrite, [replacement]))
+ finally:
+ commit.close()
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_kind == 'OVERWRITE'
+ _assert_codec(table, snapshot, codec, primary_key)
+ assert _read(table, use_native) == [expected[1], expected[3], replacement]
+
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ with _publication(server, commit, use_native):
+ commit.truncate_partitions([{'pt': 'b'}])
+ finally:
+ commit.close()
+ assert _read(table, use_native) == [expected[3], replacement]
+ _assert_codec(table, table.snapshot_manager().get_latest_snapshot(),
codec, primary_key)
+
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ with _publication(server, commit, use_native):
+ commit.truncate_table()
+ finally:
+ commit.close()
+ assert _read(table, use_native) == []
+ assert _read(table, use_native, history.id) == expected
+ assert _read(table, use_native, first.id) == _SEED
diff --git
a/paimon-python/pypaimon/tests/native_partition_delete_changelog_test.py
b/paimon-python/pypaimon/tests/native_partition_delete_changelog_test.py
new file mode 100644
index 0000000000..7952383c1d
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_partition_delete_changelog_test.py
@@ -0,0 +1,104 @@
+# 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.
+
+"""Whole-partition deletes do not use the row-level changelog writer."""
+
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.write.native_update import create_native_predicate_delete,
create_native_update
+from pypaimon.write.table_update import TableUpdate
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust is not installed'))]
+
+
+def _table(catalog, native=False, guarded=False):
+ schema = pa.schema([('id', pa.int32()), ('pt', pa.string()), ('value',
pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['pt'], primary_keys=['id', 'pt'], options={
+ 'bucket': '1', 'changelog-producer': 'input',
'write.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false',
+ 'commit.native.enabled': 'false', 'query-auth.enabled':
str(guarded).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ dict(id=1, pt='a', value=10), dict(id=2, pt='b', value=20),
+ ], schema=schema))
+ messages = writer.prepare_commit()
+ assert sum(len(message.changelog_files) for message in messages) == 2
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ return table.copy({'write.native.enabled': str(native).lower()})
+
+
+def
test_only_partition_delete_qualifies_a_native_changelog_table(native_rest_catalog):
+ table = _table(native_rest_catalog, True)
+ predicates = table.new_read_builder().new_predicate_builder()
+ native_table = MagicMock()
+ with patch('pypaimon.write.native_update.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_update.create_native_write_table',
return_value=native_table) as create:
+ assert create_native_predicate_delete(table, 'partition',
predicates.equal('pt', 'a')) is not None
+ create.assert_called_once()
+ create.reset_mock()
+ for predicate in [None, predicates.equal('id', 1),
PredicateBuilder.and_predicates([
+ predicates.equal('pt', 'a'), predicates.equal('value', 10)]),
+ Predicate('equal', 1, 'pt', ['a'], field_path=['value']),
+ Predicate('equal', 1, 'pt', ['a'], field_path=['pt',
'nested'])]:
+ assert create_native_predicate_delete(table, 'rows', predicate) is
None
+ assert create_native_update(table, 'rows', ['value']) is None
+ create.assert_not_called()
+
+ assert create_native_predicate_delete(object(), 'unsupported', None) is
None
+
+
[email protected]('native', _ENGINES)
[email protected]('guarded', [False, True])
+def
test_input_changelog_partition_delete_keeps_other_partitions(native_rest_catalog,
native, guarded):
+ table = _table(native_rest_catalog, native, guarded)
+ predicate = table.new_read_builder().new_predicate_builder().equal('pt',
'a')
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(
+ TableUpdate, '_delete_by_partition_filter',
side_effect=AssertionError('Python delete fallback')))
+ messages = builder.new_update().delete_by_predicate(predicate)
+ assert [tuple(message.partition) for message in messages] == [('a',)]
+ assert sum(len(message.deleted_files) for message in messages) == 1
+ assert all(not message.new_files and not message.changelog_files for
message in messages)
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ read = table.new_read_builder()
+ assert
read.new_read().to_arrow(read.new_scan().plan().splits()).to_pylist() == [
+ dict(id=2, pt='b', value=20)]
diff --git a/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
b/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
index f6e2e9e8b8..7813cb6140 100644
--- a/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
+++ b/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
@@ -20,14 +20,16 @@
from contextlib import ExitStack
from pathlib import Path
from urllib.parse import unquote, urlparse
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
import pyarrow as pa
import pytest
from pypaimon import Schema
from pypaimon.read.native_plan import native_runtime_available
-from pypaimon.write.native_update import _native_upsert_key_type_compatible
+from pypaimon.write.native_update import (
+ _native_upsert_key_compatible, _native_upsert_key_type_compatible,
create_native_upsert,
+)
from pypaimon.write.table_upsert_by_key import TableUpsertByKey
@@ -226,3 +228,124 @@ def
test_typed_null_union_children_match_existing_null_keys(
dict(key=None, p='a', value=91, keep='null'), dict(key=1, p='b',
value=40, keep='other-partition'),
dict(key=2, p='a', value=20, keep='second'),
]
+
+
+def _typed_null_keys(kind, value_type):
+ if kind == 'plain':
+ return pa.nulls(3, type=value_type)
+ unused = 'not an integer' if pa.types.is_string(value_type) else
{'unused': 'not an integer'}
+ values = pa.array([unused, None, unused], type=value_type)
+ if kind == 'nested_dictionary':
+ values = pa.DictionaryArray.from_arrays(pa.array([0, 1, 2],
type=pa.int8()), values)
+ return pa.DictionaryArray.from_arrays(pa.array([0, 1, None, 1, 2],
type=pa.int8()), values).slice(1, 3)
+
+
+def _typed_null_union_keys(dense, dictionary, non_null=False):
+ tags = pa.array([7, 42, 7, 42, 7], type=pa.int8())
+ middle = '1' if non_null else None
+ row_type = pa.struct([('unused', pa.string())])
+ if dense:
+ values = pa.UnionArray.from_dense(
+ tags, pa.array([0, 0, 1, 0, 2], type=pa.int32()),
+ [pa.array(['outside', middle, 'outside']),
+ pa.array([None, {'unused': 'unselected'}], type=row_type)],
type_codes=[7, 42])
+ else:
+ values = pa.UnionArray.from_sparse(
+ tags, [pa.array(['outside', 'inactive', middle, 'inactive',
'outside']),
+ pa.array([{'unused': 'inactive'}, None, {'unused':
'inactive'}, None,
+ {'unused': 'inactive'}], type=row_type)],
type_codes=[7, 42])
+ # Nonzero parent offsets must select child tags/offsets from the slice.
+ values = values.slice(1, 3)
+ if dictionary:
+ values = pa.DictionaryArray.from_arrays(
+ pa.array([0, 0, 1, None, 2], type=pa.int8()), values).slice(1, 3)
+ return values
+
+
[email protected]('dense', [False, True])
[email protected]('dictionary', [False, True])
[email protected]('non_null', [False, True])
+def test_null_union_key_gate_uses_selected_logical_values(dense, dictionary,
non_null):
+ keys = _typed_null_union_keys(dense, dictionary, non_null)
+ assert keys.to_pylist() == [None, '1' if non_null else None, None]
+ assert not _native_upsert_key_type_compatible(keys.type, pa.int32())
+ assert _native_upsert_key_compatible(keys, pa.int32()) is not non_null
+ chunks = pa.chunked_array([keys.slice(0, 1), keys.slice(1)])
+ assert _native_upsert_key_compatible(chunks, pa.int32()) is not non_null
+
+
+def test_unrelated_non_null_key_types_still_use_the_existing_python_path():
+ for source in [pa.array([None, '1']), pa.array([True]),
pa.array([{'unused': 'value'}])]:
+ assert not _native_upsert_key_compatible(source, pa.int32())
+ source = pa.DictionaryArray.from_arrays(
+ pa.array([0, 1], type=pa.int8()), pa.array([None, '1']))
+ assert not _native_upsert_key_compatible(source, pa.int32())
+ assert not _native_upsert_key_compatible(pa.chunked_array([source.slice(0,
1), source]), pa.int32())
+ # Already compatible types do not inspect value or kernel availability.
+ with patch('pypaimon.write.native_update.pc.take',
side_effect=AssertionError('unnecessary take')):
+ assert _native_upsert_key_compatible(pa.array([1], type=pa.int32()),
pa.int32())
+
+
[email protected]('kind', ['plain', 'dictionary', 'nested_dictionary'])
[email protected]('value_type', [pa.string(), pa.struct([('unused',
pa.string())])])
+def
test_all_null_key_columns_select_native_even_with_an_unrelated_physical_type(
+ native_rest_catalog, kind, value_type):
+ table = _table(native_rest_catalog, pa.int32(), [1, 2], False, True)
+ keys = _typed_null_keys(kind, value_type)
+ source = pa.table({'key': keys, 'p': ['a'] * 3, 'value': pa.array([90, 91,
92], type=pa.int32())})
+ source = pa.concat_tables([source.slice(0, 1), source.slice(1)])
+ native_table = MagicMock()
+ with patch('pypaimon.write.native_update._native_row_id_table',
return_value=native_table):
+ selected = create_native_upsert(table, 'null-keys', source, ['key'],
['value'])
+ assert selected is not None
+ native_table.new_batch_write_builder.assert_called_once()
+
+
[email protected]('native', _ENGINES)
[email protected]('kind', ['plain', 'dictionary', 'nested_dictionary'])
[email protected]('value_type', [pa.string(), pa.struct([('unused',
pa.string())])])
+def test_all_null_key_columns_update_existing_null_rows(
+ native_rest_catalog, native, kind, value_type):
+ table = _table(native_rest_catalog, pa.int32(), [1, 2], native, True)
+ keys = _typed_null_keys(kind, value_type)
+ source = pa.table({'key': keys, 'p': ['a'] * 3, 'value': pa.array([90, 91,
92], type=pa.int32())})
+ source = pa.concat_tables([source.slice(0, 1), source.slice(1)])
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, native)
+ messages =
builder.new_update().with_update_type(['value']).upsert_by_arrow_with_key(source,
['key'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert _read(table) == [
+ dict(key=1, p='a', value=11, keep='duplicate'), dict(key=1, p='a',
value=10, keep='first'),
+ dict(key=None, p='a', value=92, keep='null'), dict(key=1, p='b',
value=40, keep='other-partition'),
+ dict(key=2, p='a', value=20, keep='second'),
+ ]
+
+
[email protected]('native', _ENGINES)
[email protected]('dense', [False, True])
[email protected]('dictionary', [False, True])
+def test_all_null_union_keys_update_existing_null_rows(
+ native_rest_catalog, native, dense, dictionary):
+ table = _table(native_rest_catalog, pa.int32(), [1, 2], native, True)
+ keys = _typed_null_union_keys(dense, dictionary)
+ source = pa.table({'key': keys, 'p': ['a'] * 3, 'value': pa.array([90, 91,
92], type=pa.int32())})
+ source = pa.concat_tables([source.slice(0, 1), source.slice(1)])
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, native)
+ messages =
builder.new_update().with_update_type(['value']).upsert_by_arrow_with_key(source,
['key'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert _read(table) == [
+ dict(key=1, p='a', value=11, keep='duplicate'), dict(key=1, p='a',
value=10, keep='first'),
+ dict(key=None, p='a', value=92, keep='null'), dict(key=1, p='b',
value=40, keep='other-partition'),
+ dict(key=2, p='a', value=20, keep='second'),
+ ]
diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py
b/paimon-python/pypaimon/tests/partition_predicate_test.py
index f1d0e96fb2..0136156d30 100644
--- a/paimon-python/pypaimon/tests/partition_predicate_test.py
+++ b/paimon-python/pypaimon/tests/partition_predicate_test.py
@@ -258,7 +258,7 @@ class TestOverwritePartitionPredicate(unittest.TestCase):
with
patch('pypaimon.write.commit.overwrite_changes_provider.FileScanner') as
mock_cls:
mock_cls.return_value.read_manifest_entries.return_value = []
commit.manifest_list_manager.read_all.return_value = []
- entries_plan(Mock(id=1))
+ entries_plan(Mock(id=1, index_manifest=None))
return mock_cls.call_args[1]['partition_predicate']
def test_overwrite_rejects_mismatched_partition(self, *_):
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 8ba69b936c..adc230e51a 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -50,6 +50,7 @@ from pypaimon.write.commit.overwrite_changes_provider import
OverwriteChanges, O
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_callback import CommitCallback,
CommitCallbackContext
from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.partition_spec import _UNREPRESENTABLE_PARTITION_VALUE,
_partition_spec_value
logger = logging.getLogger(__name__)
@@ -327,30 +328,12 @@ class FileStoreCommit:
if not partitions:
raise ValueError("Partitions list cannot be empty.")
- partition_keys_set = set(self.table.partition_keys)
- for part in partitions:
- for key in part:
- if key not in partition_keys_set:
- raise ValueError(
- f"Partition spec key '{key}' is not a partition
column. "
- f"Partition keys are:
{list(self.table.partition_keys)}."
- )
-
predicate_builder = PredicateBuilder(self.table.partition_keys_fields)
- default_part_value = self.table.options.options.get(
- CoreOptions.PARTITION_DEFAULT_NAME, "__DEFAULT_PARTITION__")
partition_predicates = []
for part in partitions:
- sub_predicates = []
- for key, value in part.items():
- if value is None or (isinstance(value, str) and value ==
default_part_value):
- sub_predicates.append(predicate_builder.is_null(key))
- else:
- sub_predicates.append(predicate_builder.equal(key, value))
- if sub_predicates:
- pred = predicate_builder.and_predicates(sub_predicates)
- if pred is not None:
- partition_predicates.append(pred)
+ pred = self._create_partition_spec_filter(part)
+ if pred is not None:
+ partition_predicates.append(pred)
if not partition_predicates:
raise RuntimeError("Failed to build partition filter for
drop_partitions.")
@@ -879,24 +862,45 @@ class FileStoreCommit:
def _create_static_partition_filter(self, overwrite_partition,
commit_messages: List[CommitMessage]):
"""Build a partition filter from the explicit overwrite_partition
spec."""
- if not overwrite_partition:
+ partition_filter =
self._create_partition_spec_filter(overwrite_partition)
+ if partition_filter is None:
+ return None
+ for msg in commit_messages:
+ row = OffsetRow(msg.partition, 0, len(msg.partition))
+ if not partition_filter.test(row):
+ raise RuntimeError(f"Trying to overwrite partition
{overwrite_partition}, but the changes "
+ f"in {msg.partition} does not belong to
this partition")
+ return partition_filter
+
+ def _create_partition_spec_filter(self, partition_spec):
+ """Convert textual spec values before matching manifest rows or
stats."""
+ if not partition_spec:
return None
predicate_builder = PredicateBuilder(self.table.partition_keys_fields)
+ fields = {field.name: field for field in
self.table.partition_keys_fields}
default_part_value = self.table.options.options.get(
CoreOptions.PARTITION_DEFAULT_NAME, "__DEFAULT_PARTITION__")
equalities = []
- for key, value in overwrite_partition.items():
+ for key, value in partition_spec.items():
+ if key not in fields:
+ raise ValueError(f"Partition spec key '{key}' is not a
partition column. "
+ f"Partition keys are:
{list(self.table.partition_keys)}.")
if value is None or (isinstance(value, str) and value ==
default_part_value):
equalities.append(predicate_builder.is_null(key))
else:
- equalities.append(predicate_builder.equal(key, value))
- partition_filter = predicate_builder.and_predicates(equalities)
- for msg in commit_messages:
- row = OffsetRow(msg.partition, 0, len(msg.partition))
- if not partition_filter.test(row):
- raise RuntimeError(f"Trying to overwrite partition
{overwrite_partition}, but the changes "
- f"in {msg.partition} does not belong to
this partition")
- return partition_filter
+ try:
+ value = _partition_spec_value(value, fields[key].type)
+ except (ValueError, TypeError, OverflowError) as error:
+ raise ValueError("Invalid partition value for '%s': %r" %
(key, value)) from error
+ if value is _UNREPRESENTABLE_PARTITION_VALUE:
+ # Keep valid Java values outside Python's partition range
+ # unmatched, without wrapping a TIME into another day.
+ equalities.append(predicate_builder.and_predicates([
+ predicate_builder.is_null(key),
predicate_builder.is_not_null(key)]))
+ else:
+ equalities.append(predicate_builder.is_null(key) if value
is None
+ else predicate_builder.equal(key, value))
+ return predicate_builder.and_predicates(equalities)
def _overwrite_changes_provider(self, partition_filter, commit_messages):
"""Build a stateful provider of OVERWRITE commit entries that caches
the
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 186f5f8f53..4bc293b8bb 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -20,6 +20,7 @@
import json
import pyarrow as pa
+import pyarrow.compute as pc
from pypaimon.read.native_plan import _native_query_auth_supported,
native_runtime_available
from pypaimon.schema.arrow_schema import arrow_schemas_compatible
@@ -41,12 +42,12 @@ def _native_update_formats_supported(table):
and (not table.options.with_vector_format() or
table.options.vector_file_format() == 'parquet'))
-def _native_update_table(table):
+def _native_update_table(table, *, allow_changelog=False):
"""Resolve common native update prerequisites before preparing changes."""
if (type(table) is not FileStoreTable
or not table.options.native_write_enabled()
or not _native_update_formats_supported(table)
- or table.options.changelog_producer() != ChangelogProducer.NONE
+ or (not allow_changelog and table.options.changelog_producer() !=
ChangelogProducer.NONE)
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
or not native_runtime_available()
or not _native_query_auth_supported(table)):
@@ -205,6 +206,47 @@ def _native_upsert_key_type_compatible(source, target):
check_top_level_nullability=False, allow_binary_compatibility=True)
+def _native_upsert_key_compatible(source, target):
+ if _native_upsert_key_type_compatible(source.type, target):
+ return True
+
+ def nulls(array):
+ if pa.types.is_dictionary(array.type):
+ # Index validity alone misses a selected NULL dictionary value.
+ # Select only its validity bitmap, without decoding key payloads.
+ return pc.fill_null(pc.take(nulls(array.dictionary),
array.indices), True)
+ if pa.types.is_union(array.type):
+ # Unions have no validity bitmap. Only the active child's NULL
+ # counts; inactive children can contain unrelated non-NULL values.
+ # Read the parent buffers at its offset because PyArrow's union
+ # type_codes/offsets accessors can omit the offset of a slice.
+ buffers = array.buffers()
+ codes = pa.Array.from_buffers(pa.int8(), len(array), [None,
buffers[1]], offset=array.offset)
+ offsets = (pa.Array.from_buffers(pa.int32(), len(array), [None,
buffers[2]], offset=array.offset)
+ if array.type.mode == 'dense' else None)
+ result = pc.fill_null(pa.nulls(len(array), type=pa.bool_()), False)
+ for index, code in enumerate(array.type.type_codes):
+ active = pc.equal(codes, code)
+ child_nulls = nulls(array.field(index))
+ if offsets is not None:
+ child_nulls = pc.take(child_nulls, pc.if_else(active,
offsets, None))
+ result = pc.if_else(active, child_nulls, result)
+ return result
+ return array.is_null()
+
+ try:
+ chunks = source.chunks if isinstance(source, pa.ChunkedArray) else
[source]
+ for chunk in chunks:
+ if chunk.null_count != len(chunk) and not
pc.all(nulls(chunk)).as_py():
+ return False
+ # Keep Python's existing input contract. In particular, bare view
+ # and run-end inputs have no take kernel on supported PyArrow.
+ pc.take(chunk, pa.array([], type=pa.int64()))
+ return True
+ except (pa.ArrowInvalid, pa.ArrowNotImplementedError, TypeError):
+ return False
+
+
def _native_update_columns_supported(table, columns):
"""Core writes Parquet columns and Java-compatible sparse Blob deltas."""
if type(table) is not FileStoreTable:
@@ -272,7 +314,7 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
# surviving payloads before applying row-ID update or append rules;
# validating all payload types here would reject valid matched updates.
key_names = list(dict.fromkeys(keys + table.partition_keys))
- if not
all(_native_upsert_key_type_compatible(data.schema.field(name).type,
schema.field(name).type)
+ if not all(_native_upsert_key_compatible(data.column(name),
schema.field(name).type)
for name in key_names):
return None
writer = (native_table.new_batch_write_builder()
@@ -320,9 +362,25 @@ def native_predicate_row_ids(scan_table, predicate,
splits):
def create_native_predicate_delete(table, commit_user, predicate):
"""Delegate matching and partition metadata deletion to the core
updater."""
from pypaimon.read.native_plan import _predicate_to_native
+ if type(table) is not FileStoreTable:
+ return None
+
+ def canonical_paths(node):
+ if node.field_path and node.field_path != [node.field]:
+ return False
+ return (all(canonical_paths(child) for child in (node.literals or []))
+ if node.method in ('and', 'or') else True)
+
# Core selects whole-partition metadata deletes for append/PK tables too;
# only the row-level branch requires row tracking and deletion vectors.
- native_table = _native_update_table(table)
+ from pypaimon.write.table_update import TableUpdate
+ fields = set() if predicate is None else
TableUpdate._predicate_fields(predicate)
+ partition_delete = (table.is_primary_key_table and bool(fields)
+ and fields.issubset(table.partition_keys) and
canonical_paths(predicate))
+ # Partition deletion removes complete file/index metadata. It does not
+ # produce row-level changes, even when the PK writer emits changelogs.
+ # Core still validates the translated predicate and selects this fast path.
+ native_table = _native_update_table(table,
allow_changelog=partition_delete)
if native_table is None:
return None
native_predicate = None if predicate is None else
_predicate_to_native(predicate)
diff --git a/paimon-python/pypaimon/write/partition_spec.py
b/paimon-python/pypaimon/write/partition_spec.py
new file mode 100644
index 0000000000..177c15ddca
--- /dev/null
+++ b/paimon-python/pypaimon/write/partition_spec.py
@@ -0,0 +1,226 @@
+# 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.
+
+"""Java TypeUtils conversions for private overwrite and truncate specs."""
+
+import re
+import struct
+from calendar import monthrange
+from datetime import date, datetime, timedelta
+
+
+_UNREPRESENTABLE_PARTITION_VALUE = object()
+_JAVA_WHITESPACE = ''.join(chr(value) for value in range(33))
+
+
+def _partition_spec_value(value, data_type):
+ """TypeUtils.castFromString for the scalar types accepted as partitions.
+
+ Partition specs are not SQL casts: integer text must be integral and
+ DECIMAL uses HALF_UP with precision overflow represented as NULL.
+ Already typed Python values retain the existing public API contract.
+ """
+ if not isinstance(value, str):
+ return value
+ import pyarrow as pa
+
+ from pypaimon.casting.decimal_cast import cast_decimal
+ from pypaimon.casting.numeric_utils import java_floating_string
+ from pypaimon.casting.temporal_utils import _local_to_utc_seconds, _signed
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ from pypaimon.table.row.generic_row import _parse_type_precision_scale
+ from unicodedata import decimal
+
+ target = PyarrowFieldParser.from_paimon_type(data_type)
+ if pa.types.is_integer(target) or pa.types.is_decimal(target):
+ # Integer.valueOf and BigDecimal scan UTF-16 chars. BMP decimal
+ # digits are accepted; a supplementary-plane digit is two surrogates.
+ value = ''.join(str(decimal(char)) if ord(char) <= 0xffff and
char.isdecimal() else char for char in value)
+ if pa.types.is_string(target) or pa.types.is_binary(target):
+ result = value.encode('utf-8') if pa.types.is_binary(target) else value
+ match = re.fullmatch(r'(CHAR|VARCHAR|VARBINARY)\((\d+)\)',
data_type.type.upper())
+ if match is not None and len(result) > int(match.group(2)):
+ raise ValueError('Partition value exceeds the declared type
length')
+ return result
+ if pa.types.is_integer(target):
+ if re.fullmatch(r'[+-]?[0-9]+', value) is None:
+ raise ValueError('Partition integer text must be integral')
+ result = int(value)
+ if not -(1 << (target.bit_width - 1)) <= result < (1 <<
(target.bit_width - 1)):
+ raise ValueError('Partition integer value is out of range')
+ return result
+ if pa.types.is_boolean(target):
+ lowered = value.lower()
+ if lowered in ('t', 'true', 'y', 'yes', '1'):
+ return True
+ if lowered in ('f', 'false', 'n', 'no', '0'):
+ return False
+ raise ValueError('Invalid BOOLEAN partition value')
+ if pa.types.is_floating(target):
+ text = value.strip(_JAVA_WHITESPACE)
+ if
re.fullmatch(r'[+-]?(?:NaN|Infinity|(?:[0-9]+(?:\.[0-9]*)?|\.[0-9]+)'
+
r'(?:[eE][+-]?[0-9]+)?[fFdD]?|0[xX](?:[0-9a-fA-F]+(?:\.[0-9a-fA-F]*)?'
+ r'|\.[0-9a-fA-F]+)'
+ r'[pP][+-]?[0-9]+[fFdD]?)', text) is None:
+ raise ValueError('Invalid floating partition value')
+ text = text[:-1] if text[-1:] in ('f', 'F', 'd', 'D') else text
+ try:
+ result = float.fromhex(text) if 'p' in text.lower() else
float(text)
+ except OverflowError:
+ # Double.parseDouble saturates overflowing hexadecimal text to
+ # infinity, while Python's float.fromhex raises instead.
+ result = float('-inf') if text.startswith('-') else float('inf')
+ if target == pa.float32():
+ try:
+ single = struct.unpack('>f', struct.pack('>f', result))[0]
+ except OverflowError:
+ single = float('-inf') if result < 0 else float('inf')
+ if result != single and java_floating_string(result) !=
java_floating_string(single, True):
+ raise ValueError('Partition value cannot be cast to FLOAT due
to precision loss')
+ return single
+ return result
+ if pa.types.is_decimal(target):
+ return cast_decimal(pa.array([value]), target)[0].as_py()
+ if pa.types.is_date32(target):
+ if re.fullmatch(r'-?[0-9]+', value):
+ days = int(value)
+ if not -(1 << 31) <= days < (1 << 31):
+ raise ValueError('DATE partition value is out of range')
+ try:
+ return date(1970, 1, 1) + timedelta(days=days)
+ except OverflowError:
+ # Pure Python cannot write partition dates outside datetime's
+ # range. A valid Java spec outside that range cannot match one.
+ return _UNREPRESENTABLE_PARTITION_VALUE
+ text = value.split(' ', 1)[0] if value.find(' ') > 0 else value
+ parts = text.split('-')
+ parts = [part.strip(_JAVA_WHITESPACE) for part in parts]
+ if not 1 <= len(parts) <= 3 or any(re.fullmatch(r'[0-9]+', part) is
None for part in parts):
+ raise ValueError('Invalid DATE partition value')
+ return date(*([int(part) for part in parts] + [1] * (3 - len(parts))))
+ if pa.types.is_time(target):
+ if re.fullmatch(r'-?[0-9]+', value):
+ millis = int(value)
+ else:
+ # Java parseTime permits partial clocks and adds a signed offset.
+ match =
re.fullmatch(r'[\x00-\x20]*([0-9]+)[\x00-\x20]*(?::[\x00-\x20]*([0-9]+)[\x00-\x20]*'
+
r'(?::[\x00-\x20]*([0-9]+)(?:\.([^+\-]*))?[\x00-\x20]*)?)?'
+ r'(?:([+-])[\x00-\x20]*([0-9]+)[\x00-\x20]*'
+ r'(?::[\x00-\x20]*([0-9]+)[\x00-\x20]*)?)?',
value)
+ if match is None:
+ raise ValueError('Invalid TIME partition value')
+ hours, minutes, seconds, fraction, sign, zone_hour, zone_minute =
match.groups()
+ if zone_minute is not None and seconds is None:
+ raise ValueError('Invalid TIME partition value')
+ fraction = (fraction or '').strip(_JAVA_WHITESPACE)
+ # DateTimeUtils.parseFraction scans UTF-16 chars. A supplementary
+ # character occupies two positions, including for rounding.
+ encoded = fraction[:4].encode('utf-16-le', 'surrogatepass')
+ units = [encoded[index] | encoded[index + 1] << 8 for index in
range(0, len(encoded), 2)]
+ millis = sum((unit - ord('0') if ord('0') <= unit <= ord('9') else
0) * factor
+ for unit, factor in zip(units, (100, 10, 1)))
+ if len(units) > 3 and units[3] >= ord('5'):
+ millis += 1
+ components = [int(item or 0) for item in (hours, minutes, seconds,
zone_hour, zone_minute)]
+ if any(item >= 1 << 31 for item in components):
+ raise ValueError('TIME partition component is out of range')
+ millis += ((int(hours) * 60 + int(minutes or 0)) * 60 +
int(seconds or 0)) * 1000
+ if sign is not None:
+ millis += (1 if sign == '+' else -1) * (int(zone_hour) * 60 +
int(zone_minute or 0)) * 60000
+ millis = _signed(millis, 32)
+ if not -(1 << 31) <= millis < (1 << 31):
+ raise ValueError('TIME partition value is out of range')
+ if not 0 <= millis < 86400000:
+ return _UNREPRESENTABLE_PARTITION_VALUE
+ return (datetime(1970, 1, 1) + timedelta(milliseconds=millis)).time()
+ if pa.types.is_timestamp(target):
+ precision = _parse_type_precision_scale(data_type)[0]
+ if target.tz is None and re.fullmatch(r'-?[0-9]+', value):
+ if precision not in (0, 3, 6, 9):
+ raise ValueError('Numeric TIMESTAMP partition values require
precision 0, 3, 6 or 9')
+ ticks = int(value)
+ if not -(1 << 63) <= ticks < (1 << 63):
+ raise ValueError('TIMESTAMP partition value is out of range')
+ # Java first converts numeric seconds to signed-long millis.
+ # The multiplication wraps before it constructs the timestamp.
+ storage_type = target
+ if precision == 0:
+ ticks = _signed(ticks * 1000, 64)
+ # A wrapped product need not retain whole-second alignment.
+ # Timestamp.fromEpochMillis keeps that millisecond remainder.
+ storage_type = pa.timestamp('ms')
+ try:
+ return pa.scalar(ticks, type=storage_type).as_py()
+ except OverflowError:
+ return _UNREPRESENTABLE_PARTITION_VALUE
+ match = re.fullmatch(r'(\+?[0-9]+)-([0-9]*)-([0-9]*)'
+ r'(?:
([0-9]*):([0-9]*):([0-9]*)(?:\.([0-9]{0,9}))?)?'
+
r'(?:T([0-9]*):([0-9]*)(?::([0-9]*)(?:\.([0-9]{0,9}))?)?)?', value)
+ if match is None:
+ raise ValueError('Invalid TIMESTAMP partition value')
+ year_text, month_text, day_text, hour, minute, second, fraction,
t_hour, t_minute, t_second, t_fraction = (
+ match.groups())
+ year_digits = year_text.lstrip('+')
+ if ((not year_text.startswith('+') and len(year_digits) != 4)
+ or year_text.startswith('+') and not 5 <= len(year_digits) <=
19
+ or not 1 <= int(year_digits) <= 999999999
+ or not month_text and day_text):
+ raise ValueError('Invalid TIMESTAMP partition value')
+
+ def component(text):
+ if not text:
+ return None
+ # DateTimeFormatter's adjacent optional [MM][M] patterns can
+ # repeat a component, provided both parsed values agree.
+ if len(text) > 21 or len(text) > 2 and int(text[:2]) !=
int(text[2:]):
+ raise ValueError('Conflicting TIMESTAMP partition components')
+ return int(text[:2])
+
+ clock = [component(text) for text in (hour, minute, second)] + [
+ None if fraction is None else int(fraction.ljust(9, '0'))]
+ other = [component(text) for text in (t_hour, t_minute, t_second)] + [
+ None if t_fraction is None else int(t_fraction.ljust(9, '0'))]
+ for index, (left, right) in enumerate(zip(clock, other)):
+ if left is not None and right is not None and left != right:
+ raise ValueError('Conflicting TIMESTAMP partition components')
+ if right is not None:
+ clock[index] = right
+ if clock[0] == 24 and (clock[1] is None and clock[2] is not None
+ or clock[2] is None and clock[3] is not None):
+ raise ValueError('Invalid TIMESTAMP partition value')
+ hour, minute, second, nanos = [item or 0 for item in clock]
+ year, month, day = int(year_digits), component(month_text),
component(day_text)
+ month = 1 if month is None else month
+ day = 1 if day is None else day
+ if (not 1 <= month <= 12 or not 1 <= day <= 31 or not 0 <= hour <= 24
+ or hour == 24 and (minute or second or nanos) or not 0 <=
minute <= 59 or not 0 <= second <= 59):
+ raise ValueError('Invalid TIMESTAMP partition value')
+ if year > 9999:
+ return _UNREPRESENTABLE_PARTITION_VALUE
+ # DateTimeFormatter's default SMART resolver clamps day-of-month and
+ # accepts exactly midnight as 24:00:00 on the following date.
+ whole = datetime(year, month, min(day, monthrange(year, month)[1]),
hour % 24, minute, second)
+ if hour == 24 and month_text and day_text:
+ whole += timedelta(days=1)
+ nanos = nanos // 10 ** (9 - precision) * 10 ** (9 - precision)
+ epoch = whole - datetime(1970, 1, 1)
+ seconds = epoch.days * 86400 + epoch.seconds
+ if target.tz is not None:
+ seconds = _local_to_utc_seconds(seconds)
+ ticks_per_second = {'s': 1, 'ms': 1000, 'us': 1000000, 'ns':
1000000000}[target.unit]
+ return pa.scalar(seconds * ticks_per_second + nanos // (1000000000 //
ticks_per_second), type=target).as_py()
+ raise ValueError('Unsupported partition spec type %s' % data_type)
