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 53b77402cf [Python] Ignore inactive format options when selecting
native mutations (#10656)
53b77402cf is described below
commit 53b77402cf8b97211341ff6c8994dbfb46d31cda
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 18:49:04 2026 +0800
[Python] Ignore inactive format options when selecting native mutations
(#10656)
---
docs/docs/pypaimon/writing.md | 7 +
.../native_update_inactive_vector_option_test.py | 172 +++++++++++++++++++++
.../tests/native_write_inactive_options_test.py | 150 ++++++++++++++++++
paimon-python/pypaimon/write/native_update.py | 5 +-
paimon-python/pypaimon/write/native_write.py | 8 +-
5 files changed, 338 insertions(+), 4 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index e84209ef6c..6014a8a3b9 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -104,6 +104,13 @@ Whole-partition predicate deletes can use Native on
primary-key tables configure
with input changelogs. Native upserts also accept fully NULL source key columns
with a different physical type, including selected NULL dictionary values.
+Native selection checks the formats used by the operation. An unused
+`changelog-file.format` does not prevent Native Parquet writes when no INPUT
+changelog is produced, including batch overwrite. Likewise,
`vector.file.format`
+does not prevent Native updates on tables without VECTOR fields. These
operations
+continue to write Parquet/BLOB files; an active dedicated format must still be
+supported by the Native path.
+
`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_update_inactive_vector_option_test.py
b/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
new file mode 100644
index 0000000000..fe4ae85f7b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
@@ -0,0 +1,172 @@
+# 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.
+
+"""An unused vector format must not disable Parquet/Blob Native mutations."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.compute as pc
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.native_update import _native_update_formats_supported
+from pypaimon.write.table_delete import TableDeleteByRowId
+from pypaimon.write.table_update import TableUpdate
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_NATIVE = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_ENGINES = [False, pytest.param(True, marks=_NATIVE)]
+_SCHEMA = pa.schema([('id', pa.int32()), ('pt', pa.string()), ('value',
pa.int32()),
+ ('payload', pa.large_binary())])
+_ROWS = [dict(id=i, pt='a', value=i * 10, payload=('blob-%d' % i).encode())
for i in range(1, 4)]
+
+
[email protected]('vector,format_,expected', [
+ (False, 'orc', True), (False, 'parquet', True),
+ (True, 'orc', False), (True, 'parquet', True),
+])
+def test_update_format_gate_checks_only_active_vector_fields(tmp_path, vector,
format_, expected):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = (_SCHEMA.append(pa.field('embedding', pa.list_(pa.float32(), 2)))
if vector else _SCHEMA)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'file.format': 'parquet', 'vector.file.format': format_,
+ }), False)
+ assert _native_update_formats_supported(catalog.get_table('default.t')) is
expected
+
+
+def _write(table, rows):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ messages = writer.prepare_commit()
+ assert all(file.file_name.endswith(('.parquet', '.blob'))
+ and '.vector.' not in file.file_name for message in
messages for file in message.new_files)
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _read(table):
+ table = table.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ builder = table.new_read_builder().with_projection(table.field_names +
['_ROW_ID'])
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+
+
+def _native_only(native):
+ stack = ExitStack()
+ if native:
+ for owner, method in [(TableUpdateByRowId,
'_load_existing_files_info'),
+ (TableUpsertByKey, '_upsert_partition'),
+ (TableUpsertByKey, '_upsert_row_partition'),
+ (TableUpdate, '_build_predicate_update_table'),
+ (TableDeleteByRowId, '_scan_anchor_ranges')]:
+ stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python mutation fallback')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python MERGE
fallback')))
+ return stack
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', ['row-id', 'predicate', 'callable',
'upsert', 'row-upsert',
+ 'merge', 'merge-table', 'row-delete',
'predicate-delete'])
+def test_inactive_vector_option_preserves_parquet_blob_mutations(rest_catalog,
native, stream, operation):
+ catalog, _ = rest_catalog
+ options = {
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'deletion-vectors.enabled': 'true', 'file.format': 'parquet',
'vector.file.format': 'orc',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ }
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['pt'], options=options), False)
+ table = catalog.get_table('default.t')
+ _write(table, _ROWS)
+ before = _read(table)
+ row_id = before[0]['_ROW_ID']
+ table = table.copy({'write.native.enabled': str(native).lower(),
'commit.native.enabled': str(native).lower()})
+ source_rows = [dict(_ROWS[0], value=99), dict(id=4, pt='a', value=40,
payload=b'blob-4')]
+ source = pa.Table.from_pylist(source_rows, schema=_SCHEMA)
+ if operation == 'merge-table':
+ catalog.create_table('default.s', Schema.from_pyarrow_schema(_SCHEMA,
options=options), False)
+ source = catalog.get_table('default.s')
+ _write(source, source_rows)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ predicate = update.new_predicate_builder().equal('id', 1)
+
+ def invoke(method, *args, **kwargs):
+ if stream:
+ kwargs['commit_identifier'] = 7
+ return method(*args, **kwargs)
+
+ with _native_only(native):
+ if operation == 'row-id':
+ messages = invoke(update.update_by_arrow_with_row_id,
+ pa.table({'_ROW_ID': [row_id], 'value':
pa.array([99], type=pa.int32())}))
+ elif operation in ('predicate', 'callable'):
+ assignments = {'value': 99 if operation == 'predicate' else lambda
batch: pc.add(batch['value'], 89)}
+ messages = invoke(update.update_by_predicate, predicate,
assignments,
+ read_columns=['value'] if operation ==
'callable' else None)
+ elif operation == 'upsert':
+ messages = invoke(update.upsert_by_arrow_with_key, source, ['id'])
+ elif operation == 'row-upsert':
+ rows = [GenericRow([row[field.name] for field in table.fields],
table.fields) for row in source_rows]
+ messages = invoke(update.upsert_by_key, rows, ['id'])
+ elif operation in ('merge', 'merge-table'):
+ messages = invoke(update.merge_into, source, on=['id', 'pt'],
+ when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')])
+ elif operation == 'row-delete':
+ messages = invoke(update.delete_by_row_id, [row_id])
+ else:
+ messages = invoke(update.delete_by_predicate, predicate)
+ files = [file for message in messages for file in message.new_files]
+ assert messages
+ assert all(file.file_name.endswith(('.parquet', '.blob')) and '.vector.'
not in file.file_name for file in files)
+ commit = builder.new_commit()
+ try:
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(
+ commit.file_store_commit, 'commit',
side_effect=AssertionError('Python commit fallback')))
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ actual = [{key: value for key, value in row.items() if key != '_ROW_ID'}
for row in _read(table)]
+ expected = [dict(row) for row in _ROWS]
+ if operation.endswith('delete'):
+ expected = expected[1:]
+ else:
+ expected[0]['value'] = 99
+ if operation in ('upsert', 'row-upsert', 'merge', 'merge-table'):
+ expected.append(source_rows[1])
+ assert actual == expected
diff --git a/paimon-python/pypaimon/tests/native_write_inactive_options_test.py
b/paimon-python/pypaimon/tests/native_write_inactive_options_test.py
new file mode 100644
index 0000000000..5bb9eb19d5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_write_inactive_options_test.py
@@ -0,0 +1,150 @@
+# 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.
+
+"""Unused changelog format options do not prevent native Parquet writes."""
+
+from contextlib import ExitStack
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.write.native_write import NativeTableWrite, create_native_write
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_NATIVE = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust is not installed')
+_ENGINES = [False, pytest.param(True, marks=_NATIVE)]
+_SCHEMA = pa.schema([('id', pa.int32()), ('pt', pa.string()), ('value',
pa.int32())])
+
+
+def _table(catalog, primary_key, producer='none'):
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['pt'], primary_keys=['id', 'pt'] if
primary_key else [], options={
+ 'file.format': 'parquet', 'bucket': '1' if primary_key else '-1',
+ 'changelog-producer': producer, 'changelog-file.format': 'avro',
+ 'dynamic-partition-overwrite': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ }), False)
+ return catalog.get_table('default.t')
+
+
[email protected]('primary_key,stream,partition,producer', [
+ (False, False, None, 'none'), (False, True, None, 'none'),
+ (True, False, None, 'none'), (True, True, None, 'none'),
+ (True, False, {}, 'input'), (True, False, {'pt': 'a'}, 'input'),
+])
+def test_inactive_changelog_format_qualifies_for_native_write(
+ native_rest_catalog, primary_key, stream, partition, producer):
+ table = _table(native_rest_catalog, primary_key, producer)
+ native_table = Mock()
+ with patch('pypaimon.write.native_write.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_write.create_native_write_table',
return_value=native_table) as create:
+ writer = create_native_write(table, 'job', static_partition=partition,
stream=stream)
+ assert isinstance(writer, NativeTableWrite)
+ create.assert_called_once_with(table)
+ builder =
(native_table.new_stream_write_builder.return_value.with_commit_user.return_value
if stream
+ else
native_table.new_batch_write_builder.return_value._with_commit_user.return_value)
+ if partition is not None:
+ builder.with_overwrite.assert_called_once_with(partition)
+ builder = builder.with_overwrite.return_value
+ assert writer._native_writer is builder.new_write.return_value
+ writer.close()
+
+
[email protected]('stream,partition', [(False, None), (True, None),
(True, {})])
+def test_active_non_parquet_input_changelog_keeps_existing_fallback(
+ native_rest_catalog, stream, partition):
+ table = _table(native_rest_catalog, True, 'input')
+ with patch('pypaimon.write.native_write.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_write.create_native_write_table') as
create:
+ assert create_native_write(table, 'job', static_partition=partition,
stream=stream) is None
+ create.assert_not_called()
+
+
+def _write(table, rows, stream=False, overwrite=None):
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ if overwrite is not None:
+ builder.overwrite(overwrite)
+ native = table.options.native_write_enabled()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(
+ NativeTableWrite, '_switch_to_python',
side_effect=AssertionError('Python write fallback')))
+ writer, commit = builder.new_write(), builder.new_commit()
+ if native:
+ assert isinstance(writer, NativeTableWrite)
+ for method in ('commit', 'overwrite'):
+ stack.enter_context(patch.object(
+ commit.file_store_commit, method,
side_effect=AssertionError('Python commit fallback')))
+ try:
+ messages = []
+ # Stream checkpoints exercise reuse after the first prepared batch.
+ for identifier, group in enumerate(([row] for row in rows) if
stream else [rows], 1):
+ writer.write_arrow(pa.Table.from_pylist(group, schema=_SCHEMA))
+ current = writer.prepare_commit(identifier) if stream else
writer.prepare_commit()
+ assert current
+ assert all(file.file_name.endswith('.parquet')
+ for message in current for file in
message.new_files)
+ assert all(not message.changelog_files for message in current)
+ if stream:
+ commit.commit(current, identifier)
+ else:
+ commit.commit(current)
+ messages.extend(current)
+ finally:
+ writer.close()
+ commit.close()
+ assert
table.snapshot_manager().get_latest_snapshot().changelog_manifest_list is None
+ return messages
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+
+
[email protected]('native', _ENGINES)
[email protected]('primary_key', [False, True], ids=['append', 'pk'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
+def test_parquet_writes_ignore_unused_non_parquet_changelog_format(
+ native_rest_catalog, native, primary_key, stream):
+ table = _table(native_rest_catalog, primary_key).copy({
+ 'write.native.enabled': str(native).lower(), 'commit.native.enabled':
str(native).lower()})
+ rows = [dict(id=1, pt='a', value=10), dict(id=2, pt='b', value=20)]
+ _write(table, rows, stream)
+ assert _rows(table) == rows
+
+
[email protected]('native', _ENGINES)
[email protected]('primary_key', [False, True], ids=['append', 'pk'])
[email protected]('partition', [{}, {'pt': 'a'}], ids=['all',
'partition'])
+def test_overwrite_suppresses_inactive_non_parquet_input_changelog(
+ native_rest_catalog, native, primary_key, partition):
+ table = _table(native_rest_catalog, primary_key)
+ seed = [dict(id=1, pt='a', value=10), dict(id=2, pt='b', value=20)]
+ _write(table, seed)
+ table = table.copy({'changelog-producer': 'input' if primary_key else
'none',
+ 'write.native.enabled': str(native).lower(),
+ 'commit.native.enabled': str(native).lower()})
+ replacement = [dict(id=3, pt='a', value=30)]
+ _write(table, replacement, overwrite=partition)
+ expected = ([seed[1]] if partition else []) + replacement
+ assert _rows(table) == sorted(expected, key=lambda row: row['id'])
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 4bc293b8bb..a7825c13dd 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -24,7 +24,7 @@ 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
-from pypaimon.schema.data_types import MapType, PyarrowFieldParser, RowType,
is_blob_file_field
+from pypaimon.schema.data_types import MapType, PyarrowFieldParser, RowType,
VectorType, is_blob_file_field
from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
@@ -39,7 +39,8 @@ from pypaimon.write.row_utils import _contains_blob_value,
value_for_arrow
def _native_update_formats_supported(table):
return (table.options.file_format() == 'parquet'
- and (not table.options.with_vector_format() or
table.options.vector_file_format() == 'parquet'))
+ and (not table.options.with_vector_format() or
table.options.vector_file_format() == 'parquet'
+ or not any(isinstance(field.type, VectorType) for field in
table.table_schema.fields)))
def _native_update_table(table, *, allow_changelog=False):
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 267694d695..ce74056e24 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -18,7 +18,7 @@
import pyarrow as pa
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
from pypaimon.read.native_plan import native_runtime_available
from pypaimon.schema.arrow_schema import (
arrow_schemas_compatible, normalize_arrow_strings,
normalize_arrow_write_nullability)
@@ -51,7 +51,11 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
BucketMode.BUCKET_UNAWARE,
BucketMode.CROSS_PARTITION,
BucketMode.POSTPONE_MODE)
- or table.options.changelog_file_format() not in (None, 'parquet')
+ # The format matters only when this write emits INPUT changelogs.
+ # Batch overwrites suppress them because they cannot be committed.
+ or (table.options.changelog_file_format() not in (None, 'parquet')
+ and table.options.changelog_producer() ==
ChangelogProducer.INPUT
+ and (stream or static_partition is None))
or table.options.file_format() != 'parquet'
or not _native_partition_types_supported(
schema, table.partition_keys,
table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME))
