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 c60b7a93e2 [Python] Enable native Blob and VARIANT updates with Java
layouts (#10390)
c60b7a93e2 is described below
commit c60b7a93e25fa6c4dedcb46f8bed5452fd51ed09
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 5 19:03:44 2026 +0800
[Python] Enable native Blob and VARIANT updates with Java layouts (#10390)
---
.../pypaimon/common/options/core_options.py | 29 +-
.../pypaimon/read/reader/format_pyarrow_reader.py | 5 +-
paimon-python/pypaimon/table/file_store_table.py | 24 +-
paimon-python/pypaimon/tests/blob_table_test.py | 20 +-
paimon-python/pypaimon/tests/blob_test.py | 8 +-
.../pypaimon/tests/native_blob_update_test.py | 330 +++++++++++++++++++++
.../pypaimon/tests/native_variant_update_test.py | 195 ++++++++++++
.../table/time_travel_schema_evolution_test.py | 28 +-
.../tests/table_update_by_row_id_chunked_test.py | 4 +-
paimon-python/pypaimon/write/native_update.py | 38 ++-
paimon-python/pypaimon/write/row_utils.py | 17 +-
paimon-python/pypaimon/write/table_update.py | 2 +-
.../pypaimon/write/table_update_by_row_id.py | 17 +-
paimon-python/pypaimon/write/writer/data_writer.py | 3 +-
.../write/writer/dedicated_format_writer.py | 23 +-
15 files changed, 646 insertions(+), 97 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index e419a70415..525b16f708 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -1199,18 +1199,6 @@ class CoreOptions:
)
)
- VARIANT_SHREDDING_ENABLED: ConfigOption[bool] = (
- ConfigOptions.key("variant.shredding.enabled")
- .boolean_type()
- .default_value(True)
- .with_description(
- "Whether to enable VARIANT shredding. When True (default), writes
apply the "
- "shredding schema configured via 'variant.shreddingSchema', and
reads "
- "automatically reassemble shredded columns back to the standard "
- "struct<value, metadata> form. Set to False to bypass both
behaviours."
- )
- )
-
VARIANT_SHREDDING_SCHEMA: ConfigOption[str] = (
ConfigOptions.key("variant.shreddingSchema")
.string_type()
@@ -1398,9 +1386,6 @@ class CoreOptions:
def blob_as_descriptor(self, default=None):
return self.options.get(CoreOptions.BLOB_AS_DESCRIPTOR, default)
- def variant_shredding_enabled(self) -> bool:
- return self.options.get(CoreOptions.VARIANT_SHREDDING_ENABLED, True)
-
def variant_shredding_schema(self) -> Optional[str]:
val = self.options.get(CoreOptions.VARIANT_SHREDDING_SCHEMA)
if val is None:
@@ -1409,15 +1394,11 @@ class CoreOptions:
return val
def blob_descriptor_fields(self, default=None):
- # Do not treat blob.stored-descriptor-fields as a layout switch.
- # Python master ignored that key and wrote dedicated .blob payloads;
- # a global fallback would mis-parse those files during a rolling
- # upgrade. The cost is that Java tables which only set the fallback
- # key store inline descriptors, and Python returns those bytes
- # instead of fetching payload. Migrate explicitly to
- # blob-descriptor-field (column directives already copy the legacy
- # key onto the canonical option).
- value = self.options.get(CoreOptions.BLOB_DESCRIPTOR_FIELD, default)
+ # Java's fallback key applies only when the canonical key is absent.
+ if self.options.contains_key(CoreOptions.BLOB_DESCRIPTOR_FIELD.key()):
+ value = self.options.get(CoreOptions.BLOB_DESCRIPTOR_FIELD,
default)
+ else:
+ value = self.options.to_map().get('blob.stored-descriptor-fields',
default)
return CoreOptions._parse_field_set(value)
def blob_view_fields(self, default=None):
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index ecb7db7e74..58d481ccf6 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -468,8 +468,6 @@ class FormatPyArrowReader(RecordBatchReader):
self.existing_fields = [f.name for f in read_fields if f.name in
file_schema_names]
self.missing_fields = [f.name for f in read_fields if f.name not
in file_schema_names]
- self._variant_shredding_enabled = (
- options is None or options.variant_shredding_enabled())
self._variant_schema_cache: Dict[pa.DataType, VariantSchema] = {}
self._shared_shredding_maps = {}
self._selected_key_maps = {}
@@ -817,8 +815,7 @@ class FormatPyArrowReader(RecordBatchReader):
if self._selected_key_maps:
batch = self._assemble_selected_key_maps(batch)
- if self._variant_shredding_enabled:
- batch = self._assemble_shredded_variants(batch)
+ batch = self._assemble_shredded_variants(batch)
if not self.missing_fields:
return batch
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 85c9c5775b..26bad0deda 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -613,21 +613,19 @@ class FileStoreTable(Table):
historical_schema: TableSchema, options: Options,
dynamic_option_keys: Set[str]) -> dict:
# Keep scan and runtime options, but restore column declarations to
match historical fields.
historical_options = dict(options.to_map())
- for key in (
- CoreOptions.VECTOR_FIELD.key(),
- CoreOptions.BLOB_FIELD.key(),
- CoreOptions.BLOB_DESCRIPTOR_FIELD.key(),
- CoreOptions.BLOB_VIEW_FIELD.key(),
- # Restore the legacy key verbatim, not as a canonical
descriptor option:
- # Python intentionally ignores it when choosing the read
layout.
- 'blob.stored-descriptor-fields'):
- if key in dynamic_option_keys:
+ for keys in (
+ (CoreOptions.VECTOR_FIELD.key(),),
+ (CoreOptions.BLOB_FIELD.key(),),
+ (CoreOptions.BLOB_DESCRIPTOR_FIELD.key(),
'blob.stored-descriptor-fields'),
+ (CoreOptions.BLOB_VIEW_FIELD.key(),)):
+ if any(key in dynamic_option_keys for key in keys):
# Preserve explicit overrides and removals, including those
from earlier copies.
continue
- if key in historical_schema.options:
- historical_options[key] = historical_schema.options[key]
- else:
- historical_options.pop(key, None)
+ for key in keys:
+ if key in historical_schema.options:
+ historical_options[key] = historical_schema.options[key]
+ else:
+ historical_options.pop(key, None)
return historical_options
def _create_external_paths(self) -> List[str]:
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 75b6aa405b..41b86fd57b 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -5354,14 +5354,10 @@ class DedicatedFormatWriterTest(unittest.TestCase):
self.assertEqual(result['id'], list(range(2000)))
self.assertEqual(result['name'], ['updated'] * 2000)
- def test_legacy_stored_descriptor_fields_keeps_dedicated_blob_layout(self):
- """blob.stored-descriptor-fields must not switch Python to inline
descriptors.
-
- Master ignored that key and wrote dedicated .blob payloads. Head write
- with the same option must keep that layout so old readers still see
- payloads, and head read must not fail-fast on those bytes.
- """
+ def test_stored_descriptor_fallback_uses_java_inline_layout(self):
+ """The fallback option stores references in Parquet, as Java does."""
from pypaimon import Schema
+ from pypaimon.table.row.blob import BlobDescriptor
pa_schema = pa.schema([
('id', pa.int32()),
@@ -5380,11 +5376,15 @@ class DedicatedFormatWriterTest(unittest.TestCase):
table =
self.catalog.get_table('test_db.legacy_stored_descriptor_fields')
payload = b'legacy-dedicated-blob-payload'
+ source = os.path.join(self.temp_dir, 'fallback-descriptor-source')
+ with open(source, 'wb') as stream:
+ stream.write(payload)
+ descriptor = BlobDescriptor(source, 0, len(payload)).serialize()
write_builder = table.new_batch_write_builder()
writer = write_builder.new_write()
writer.write_arrow(pa.Table.from_pydict({
'id': [1],
- 'picture': [payload],
+ 'picture': [descriptor],
}, schema=pa_schema))
commit_messages = writer.prepare_commit()
write_builder.new_commit().commit(commit_messages)
@@ -5392,8 +5392,8 @@ class DedicatedFormatWriterTest(unittest.TestCase):
all_files = [f for msg in commit_messages for f in msg.new_files]
blob_files = [f for f in all_files if f.file_name.endswith('.blob')]
- self.assertGreaterEqual(len(blob_files), 1)
- self.assertTrue(all(f.write_cols == ['picture'] for f in blob_files))
+ self.assertEqual(blob_files, [])
+ self.assertTrue(all(f.file_name.endswith('.parquet') for f in
all_files))
result = table.new_read_builder().new_read().to_arrow(
table.new_read_builder().new_scan().plan().splits())
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index f08eab83df..a15f7c5ac6 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -1597,14 +1597,12 @@ class BlobTest(unittest.TestCase):
BlobDescriptor("file:///tmp/blob.bin", 0, 10),
)
- def test_blob_descriptor_fields_ignores_legacy_stored_key(self):
+ def test_blob_descriptor_fields_uses_java_fallback_key(self):
from pypaimon.common.options.core_options import CoreOptions
- # Python master ignored this key and wrote dedicated .blob files.
- # A global fallback would break rolling upgrades.
legacy_only = CoreOptions(
Options({"blob.stored-descriptor-fields": "legacy_col"}))
- self.assertEqual(set(), legacy_only.blob_descriptor_fields())
+ self.assertEqual({'legacy_col'}, legacy_only.blob_descriptor_fields())
canonical_wins = CoreOptions(Options({
"blob-descriptor-field": "canon",
@@ -2428,7 +2426,7 @@ class BlobTest(unittest.TestCase):
self.assertFalse(needs_blob_inline_convert(_Table({
"blob-as-descriptor": "true",
})))
- self.assertFalse(needs_blob_inline_convert(_Table({
+ self.assertTrue(needs_blob_inline_convert(_Table({
"blob.stored-descriptor-fields": "picture",
})))
diff --git a/paimon-python/pypaimon/tests/native_blob_update_test.py
b/paimon-python/pypaimon/tests/native_blob_update_test.py
new file mode 100644
index 0000000000..d7e9ffdc7b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_blob_update_test.py
@@ -0,0 +1,330 @@
+# 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.
+
+"""Native updates of Parquet columns preserve dedicated Blob files."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+from urllib.parse import urlparse
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.schema.data_types import AtomicType, DataField,
PyarrowFieldParser
+from pypaimon.table.row.blob import Blob, BlobDescriptor, BlobViewStruct
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.table_delete import TableDeleteByRowId
+from pypaimon.write.native_update import NativeTableUpdateByRowId
+from pypaimon.write.native_write import NativeTableWrite
+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
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, extra=None, catalog=None, payloads=None):
+ catalog = catalog or CatalogFactory.create({'warehouse': str(tmp_path /
'warehouse')})
+ catalog.create_database('db', True)
+ fields = [DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'pt', AtomicType('STRING')),
+ DataField(2, 'value', AtomicType('INT')),
+ DataField(3, 'payload', AtomicType('BLOB'))]
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ 'scan.native-plan.enabled': 'true', 'deletion-vectors.enabled':
'true',
+ 'blob.target-file-size': '16 B', 'target-file-row-num': '2'}
+ options.update(extra or {})
+ catalog.create_table('db.t', Schema(fields, partition_keys=['pt'],
options=options), False)
+ table = catalog.get_table('db.t')
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.table({'id': [0, 1, 2, 3], 'pt': ['a', 'a', 'b',
'b'],
+ 'value': [10, 11, 12, 13],
+ 'payload': payloads if payloads is not
None else
+ [b'large' * 10, None, b'', b'last']},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, schema
+
+
+def _read(table, descriptors=False, projection=None, resolve_views=True):
+ table = table.copy({'blob-as-descriptor': str(descriptors).lower(),
+ 'blob-view.resolve.enabled':
str(resolve_views).lower()})
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder.with_projection(projection)
+ splits = builder.new_scan().plan().splits()
+ read = builder.new_read()
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read fallback')):
+ return read.to_arrow(splits).sort_by('id').to_pylist()
+
+
+def _native_only():
+ stack = ExitStack()
+ for cls, method in [(TableUpdateByRowId, '_load_existing_files_info'),
+ (TableUpdate, '_build_predicate_update_table'),
+ (TableUpsertByKey, '_upsert_partition'),
+ (TableUpsertByKey, '_upsert_row_partition'),
+ (TableDeleteByRowId, 'delete')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
+ return stack
+
+
+def _commit(builder, messages, stream):
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+
+
[email protected]('stream', [False, True])
[email protected]('operation', ['row_id', 'predicate'])
+def test_native_updates_keep_blob_references(tmp_path, stream, operation):
+ table, _ = _table(tmp_path)
+ before = _read(table, descriptors=True)
+ assert BlobDescriptor.is_blob_descriptor(before[0]['payload'])
+ row_ids = {row['id']: row['_ROW_ID']
+ for row in _read(table, projection=['id', '_ROW_ID'])}
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ data = pa.table({'_ROW_ID': [row_ids[0], row_ids[2]],
+ 'value': pa.array([100, 200], type=pa.int32())})
+ with _native_only():
+ if operation == 'row_id':
+ messages = (update.update_by_arrow_with_row_id(data, 7) if stream
+ else update.update_by_arrow_with_row_id(data))
+ else:
+ predicate = PredicateBuilder(table.fields).is_in('id', [0, 2])
+
+ def assignment(matched):
+ return pa.array([100 if value == 0 else 200
+ for value in matched['id'].to_pylist()],
type=pa.int32())
+
+ args = (predicate, {'value': assignment})
+ messages = (update.update_by_predicate(*args, 7,
read_columns=['id']) if stream
+ else update.update_by_predicate(*args,
read_columns=['id']))
+ _commit(builder, messages, stream)
+ after = _read(table, descriptors=True)
+ assert [row['payload'] for row in after] == [row['payload'] for row in
before]
+ assert [row['value'] for row in after] == [100, 11, 200, 13]
+ assert [row['payload'] for row in _read(table)] == [b'large' * 10, None,
b'', b'last']
+
+
[email protected]('stream', [False, True])
[email protected]('row_input', [False, True])
+def test_native_upsert_updates_normal_columns_and_appends_blob_rows(tmp_path,
stream, row_input):
+ table, schema = _table(tmp_path)
+ before = _read(table, descriptors=True)
+ source = pa.table({'id': [0, 4], 'pt': ['a', 'b'], 'value': [100, 400],
+ 'payload': [b'ignored matched payload', b'new
payload']}, schema=schema)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ with _native_only():
+ if row_input:
+ rows = [GenericRow([row[field.name] for field in table.fields],
table.fields)
+ for row in source.to_pylist()]
+ messages = (update.upsert_by_key(rows, ['id'], 7) if stream
+ else update.upsert_by_key(rows, ['id']))
+ else:
+ messages = (update.upsert_by_arrow_with_key(source, ['id'], 7) if
stream
+ else update.upsert_by_arrow_with_key(source, ['id']))
+ _commit(builder, messages, stream)
+ rows = _read(table)
+ assert [row['value'] for row in rows] == [100, 11, 12, 13, 400]
+ assert [row['payload'] for row in rows] == [b'large' * 10, None, b'',
b'last', b'new payload']
+ after = _read(table, descriptors=True)
+ assert [row['payload'] for row in after[:4]] == [row['payload'] for row in
before]
+
+
[email protected]('stream', [False, True])
[email protected]('predicate_delete', [False, True])
+def test_native_blob_table_delete_preserves_remaining_rows(tmp_path, stream,
predicate_delete):
+ table, _ = _table(tmp_path)
+ before = _read(table, descriptors=True)
+ ids = {row['id']: row['_ROW_ID'] for row in _read(table, projection=['id',
'_ROW_ID'])}
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ with _native_only():
+ if predicate_delete:
+ predicate = PredicateBuilder(table.fields).is_in('id', [0, 2])
+ messages = (update.delete_by_predicate(predicate, 7) if stream
+ else update.delete_by_predicate(predicate))
+ else:
+ row_ids = [ids[0], ids[2], ids[0]]
+ messages = (update.delete_by_row_id(row_ids, 7) if stream
+ else update.delete_by_row_id(row_ids))
+ assert messages and all(not message.new_files for message in messages)
+ _commit(builder, messages, stream)
+ assert _read(table, descriptors=True) == [before[1], before[3]]
+ assert [row['payload'] for row in _read(table)] == [None, b'last']
+
+
+def test_normal_column_update_does_not_open_dedicated_blobs(tmp_path):
+ table, _ = _table(tmp_path)
+ ids = _read(table, projection=['id', '_ROW_ID'])
+ for path in (tmp_path / 'warehouse').rglob('*.blob'):
+ path.unlink()
+ builder = table.new_batch_write_builder()
+ with _native_only():
+ messages = builder.new_update().update_by_arrow_with_row_id(pa.table({
+ '_ROW_ID': [row['_ROW_ID'] for row in ids],
+ 'value': pa.array([100, 101, 102, 103], type=pa.int32()),
+ }))
+ assert all(file.write_cols == ['value'] for message in messages for file
in message.new_files)
+ _commit(builder, messages, False)
+ assert _read(table, projection=['id', 'value']) == [
+ {'id': index, 'value': 100 + index} for index in range(4)]
+
+
+def _inline_table(tmp_path, option, catalog=None):
+ source = tmp_path / 'references'
+ source.write_bytes(b'firstlast')
+ references = [BlobDescriptor(source.as_uri(), 0, 5).serialize(), None,
+ BlobDescriptor(source.as_uri(), 5, 0).serialize(),
+ BlobDescriptor(source.as_uri(), 5, 4).serialize()]
+ if option == 'blob-view-field':
+ references = [BlobViewStruct('db.absent_upstream', 3,
row_id).serialize()
+ if value is not None else None for row_id, value in
enumerate(references)]
+ table, schema = _table(tmp_path, {option: 'payload'}, catalog=catalog,
payloads=references)
+ return table, schema, references
+
+
[email protected]('stream', [False, True])
[email protected]('native', [False, True])
[email protected]('option', ['blob-descriptor-field',
'blob.stored-descriptor-fields', 'blob-view-field'])
[email protected]('operation', ['row_id', 'incremental', 'row', 'rows',
'predicate', 'upsert'])
+def test_native_inline_blob_updates_preserve_references(
+ tmp_path, native_rest_catalog, stream, native, option, operation):
+ # REST views reference an absent table. Updating the local reference must
+ # neither resolve it nor read an old upstream payload for unmatched rows.
+ table, schema, references = _inline_table(tmp_path, option,
native_rest_catalog)
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ ids = {row['id']: row['_ROW_ID'] for row in _read(table, projection=['id',
'_ROW_ID'])}
+ replacement = references[0]
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['payload'])
+ data = pa.table({'_ROW_ID': [ids[1], ids[2]],
+ 'payload': pa.array([replacement, None],
type=pa.large_binary())})
+ with _native_only() if native else ExitStack():
+ if operation == 'row_id':
+ messages = (update.update_by_arrow_with_row_id(data, 7) if stream
+ else update.update_by_arrow_with_row_id(data))
+ elif operation == 'incremental':
+ updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId) == native
+ updater.update_columns(data.slice(0, 1), ['payload'])
+ updater.update_columns(data.slice(1), ['payload'])
+ messages = updater.commit_messages
+ elif operation in ('row', 'rows'):
+ updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId) == native
+ value = (Blob.from_view(BlobViewStruct.deserialize(replacement))
if option == 'blob-view-field'
+ else Blob.from_descriptor_bytes(replacement,
file_io=table.file_io))
+ rows = [GenericRow([1, 'a', 999, value], table.fields),
+ GenericRow([2, 'b', 999, None], table.fields)]
+ if operation == 'row':
+ updater.update_row_columns(rows[0], [ids[1]], ['payload'])
+ updater.update_row_columns(rows[1], [ids[2]], ['payload'])
+ else:
+ updater.update_rows_columns(rows, [[ids[1]], [ids[2]]],
['payload'])
+ messages = updater.commit_messages
+ elif operation == 'predicate':
+ predicate = PredicateBuilder(table.fields).is_in('id', [1, 2])
+
+ def assignment(matched):
+ return pa.array([replacement if row_id == 1 else None
+ for row_id in matched['id'].to_pylist()],
type=pa.large_binary())
+
+ args = (predicate, {'payload': assignment})
+ messages = (update.update_by_predicate(*args, 7,
read_columns=['id']) if stream
+ else update.update_by_predicate(*args,
read_columns=['id']))
+ else:
+ source = pa.table({'id': [1, 2], 'pt': ['a', 'b'], 'value': [999,
999],
+ 'payload': [replacement, None]}, schema=schema)
+ messages = (update.upsert_by_arrow_with_key(source, ['id'], 7) if
stream
+ else update.upsert_by_arrow_with_key(source, ['id']))
+ assert all(file.file_name.endswith('.parquet') and file.write_cols ==
['payload']
+ for message in messages for file in message.new_files)
+ _commit(builder, messages, stream)
+ rows = _read(table, descriptors=True, resolve_views=False)
+ assert [row['payload'] for row in rows] == [references[0], replacement,
None, references[3]]
+ assert [row['value'] for row in rows] == [10, 11, 12, 13]
+ if option != 'blob-view-field':
+ assert [row['payload'] for row in _read(table)] == [b'first',
b'first', None, b'last']
+
+
+def test_raw_blob_incremental_updater_keeps_one_implementation(tmp_path):
+ # Later calls choose their columns independently. Keeping this operation
+ # on Python permits a scalar update followed by a raw Blob update.
+ table, _ = _table(tmp_path)
+ from pypaimon.write.native_update import create_native_update_by_row_id
+ assert create_native_update_by_row_id(table, 'test', 7) is None
+
+
+def test_view_prescan_keeps_inline_descriptor_predicate(tmp_path,
native_rest_catalog):
+ source, _ = _table(tmp_path, catalog=native_rest_catalog)
+ expected = _read(source, descriptors=True)[0]['payload']
+ row_id = _read(source, projection=['id', '_ROW_ID'])[0]['_ROW_ID']
+ file = tmp_path / 'descriptor'
+ file.write_bytes(b'unopened')
+ reference = BlobDescriptor(file.as_uri(), 0, 8).serialize()
+ fields = [DataField(0, 'id', AtomicType('INT')), DataField(1, 'value',
AtomicType('INT')),
+ DataField(2, 'reference', AtomicType('BLOB')), DataField(3,
'payload', AtomicType('BLOB'))]
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'blob-descriptor-field': 'reference', 'blob-view-field':
'payload',
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ 'scan.native-plan.enabled': 'true', 'blob-as-descriptor':
'true'}
+ native_rest_catalog.create_table('db.mixed', Schema(fields,
options=options), False)
+ table = native_rest_catalog.get_table('db.mixed')
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.table({'id': [0, 1], 'value': [10, 11],
+ 'reference': [reference, None],
+ 'payload': [BlobViewStruct('db.t', 3,
row_id).serialize(), None]}, schema=schema))
+ _commit(builder, writer.prepare_commit(), False)
+ finally:
+ writer.close()
+ with _native_only():
+ messages = builder.new_update().update_by_arrow_with_row_id(pa.table({
+ '_ROW_ID': [0], 'value': pa.array([100], type=pa.int32())}))
+ _commit(builder, messages, False)
+ for native in (False, True):
+ view = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ read_builder = view.new_read_builder().with_projection(['id',
'payload']).with_limit(1)
+
read_builder.with_filter(PredicateBuilder(view.fields).equal('reference',
reference))
+ plan = read_builder.new_scan().plan()
+ read = read_builder.new_read()
+ with (patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read fallback'))
+ if native else ExitStack()):
+ rows = read.to_arrow(plan.splits()).to_pylist()
+ assert len(rows) == 1 and rows[0]['id'] == 0
+ actual = BlobDescriptor.deserialize(rows[0]['payload'])
+ original = BlobDescriptor.deserialize(expected)
+ assert (urlparse(actual.uri).path, actual.offset, actual.length)
== (
+ urlparse(original.uri).path, original.offset, original.length)
diff --git a/paimon-python/pypaimon/tests/native_variant_update_test.py
b/paimon-python/pypaimon/tests/native_variant_update_test.py
new file mode 100644
index 0000000000..de2321172b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_variant_update_test.py
@@ -0,0 +1,195 @@
+# 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.
+
+"""Native partial updates retain Java's shredded VARIANT physical layout."""
+
+import json
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.write.native_update import NativeTableUpdateByRowId
+from pypaimon.write.native_write import NativeTableWrite
+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
+
+pytestmark = pytest.mark.native_plan
+
+
+def _variants(values):
+ return GenericVariant.to_arrow_array([
+ GenericVariant.from_python(value) if value is not None else None for
value in values])
+
+
+def _table(tmp_path, key, enabled, native_write=True):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('payload', _variants([]).type),
+ ('unchanged', _variants([]).type), ('value',
pa.int32())])
+ shredding = json.dumps({'type': 'ROW', 'fields': [
+ {'name': name, 'type': {'type': 'ROW', 'fields': [
+ {'name': 'count', 'type': 'BIGINT'}, {'name': 'name', 'type':
'STRING'},
+ ]}} for name in ('payload', 'unchanged')
+ ]})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native_write).lower(),
'target-file-row-num': '2',
+ # This Python-only option does not exist in Java. It must not disable
+ # a configured physical schema or reassembly on either reader.
+ 'variant.shredding.enabled': str(enabled).lower(), key: shredding,
+ }), False)
+ table = catalog.get_table('default.t')
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ original = [{'count': i, 'name': 'old', 'overflow': [i, None]} for i in
range(4)]
+ unchanged = [{'count': i + 10, 'name': 'untouched', 'extra': True} for i
in range(4)]
+ data = pa.table({'id': [0, 1, 2, 3], 'payload': _variants(original),
+ 'unchanged': _variants(unchanged), 'value': [10, 11, 12,
13]}, schema=schema)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite) == native_write
+ try:
+ # Rolling occurs at batch boundaries. Keep the two incremental calls
+ # in distinct file groups, as required by the public updater contract.
+ writer.write_arrow(data.slice(0, 2))
+ writer.write_arrow(data.slice(2))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, schema, original, unchanged
+
+
+def _read(table, native, projection=None):
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder.with_projection(projection)
+ splits = builder.new_scan().plan().splits()
+ read = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ data = read.to_arrow(splits).sort_by('id')
+ rows = data.to_pylist()
+ for row in rows:
+ for name in ('payload', 'unchanged'):
+ if name in row and row[name] is not None:
+ row[name] =
GenericVariant.from_arrow_struct(row[name]).to_python()
+ return rows
+
+
+def _native_only():
+ stack = ExitStack()
+ for cls, method in [(TableUpdateByRowId, '_load_existing_files_info'),
+ (TableUpdate, '_build_predicate_update_table'),
+ (TableUpsertByKey, '_upsert_partition')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
+ return stack
+
+
+def test_python_configured_variant_schema_uses_java_semantics(tmp_path):
+ table, _, original, unchanged = _table(tmp_path,
'variant.shreddingSchema', False, native_write=False)
+ for native in (False, True):
+ rows = _read(table, native)
+ assert [row['payload'] for row in rows] == original
+ assert [row['unchanged'] for row in rows] == unchanged
+ files = table.new_read_builder().new_scan().plan().splits()[0].files
+ for file in files:
+ physical = pq.read_schema(file.file_path)
+ assert 'typed_value' in [field.name for field in
physical.field('payload').type]
+
+
[email protected]('stream', [False, True])
[email protected]('key,enabled', [
+ ('variant.shreddingSchema', True), ('parquet.variant.shreddingSchema',
True),
+ ('variant.shreddingSchema', False),
+])
[email protected]('operation', ['row_id', 'incremental', 'predicate',
'upsert', 'ordinary_column'])
+def test_native_variant_partial_updates(tmp_path, stream, key, enabled,
operation):
+ table, schema, original, untouched = _table(tmp_path, key, enabled)
+ ids = {row['id']: row['_ROW_ID'] for row in _read(table, True, ['id',
'_ROW_ID'])}
+ replacement = {'count': 100, 'name': 'new', 'overflow': {'nested': [1,
None]}}
+ changed = _variants([replacement, None])
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['payload'])
+ data = pa.table({'_ROW_ID': [ids[0], ids[2]], 'payload': changed})
+ with _native_only():
+ if operation == 'row_id':
+ messages = (update.update_by_arrow_with_row_id(data, 7) if stream
+ else update.update_by_arrow_with_row_id(data))
+ elif operation == 'incremental':
+ updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId)
+ updater.update_columns(data.slice(0, 1), ['payload'])
+ updater.update_columns(data.slice(1), ['payload'])
+ messages = updater.commit_messages
+ elif operation == 'predicate':
+ predicate = PredicateBuilder(table.fields).is_in('id', [0, 2])
+
+ def assignment(matched):
+ return _variants([replacement if index == 0 else None
+ for index in matched['id'].to_pylist()])
+
+ args = (predicate, {'payload': assignment})
+ messages = (update.update_by_predicate(*args, 7,
read_columns=['id']) if stream
+ else update.update_by_predicate(*args,
read_columns=['id']))
+ elif operation == 'ordinary_column':
+ update = builder.new_update().with_update_type(['value'])
+ data = data.drop(['payload']).append_column('value',
pa.array([100, 200], type=pa.int32()))
+ messages = (update.update_by_arrow_with_row_id(data, 7) if stream
+ else update.update_by_arrow_with_row_id(data))
+ else:
+ source = pa.table({'id': [0, 2, 4], 'payload':
_variants([replacement, None, [1, 2]]),
+ 'unchanged': _variants([None, None, {'count':
14, 'name': 'untouched'}]),
+ 'value': [999, 999, 14]}, schema=schema)
+ messages = (update.upsert_by_arrow_with_key(source, ['id'], 7) if
stream
+ else update.upsert_by_arrow_with_key(source, ['id']))
+ assert messages
+ for message in messages:
+ for file in message.new_files:
+ physical = pq.read_schema(file.file_path)
+ if 'payload' in physical.names:
+ names = [field.name for field in
physical.field('payload').type]
+ assert 'typed_value' in names
+ if file.write_cols == ['value']:
+ assert physical.names == ['value']
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ expected_payload = original if operation == 'ordinary_column' else
[replacement, original[1], None, original[3]]
+ expected_values = [100, 11, 200, 13] if operation == 'ordinary_column'
else [10, 11, 12, 13]
+ if operation == 'upsert':
+ expected_payload = expected_payload + [[1, 2]]
+ expected_values += [14]
+ untouched = untouched + [{'count': 14, 'name': 'untouched'}]
+ for native in (False, True):
+ rows = _read(table, native)
+ assert [row['payload'] for row in rows] == expected_payload
+ assert [row['unchanged'] for row in rows] == untouched
+ assert [row['value'] for row in rows] == expected_values
diff --git
a/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
b/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
index 73fe088c29..17ba35de6a 100644
--- a/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
+++ b/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
@@ -143,15 +143,18 @@ class TestTimeTravelSchemaEvolution:
assert 'vector-field' not in historical.table_schema.options
assert self._read(historical) == [{'id': 1}]
- def test_legacy_descriptor_option_keeps_historical_blob_layout(self):
+ def test_descriptor_fallback_keeps_historical_inline_layout(self):
legacy_key = 'blob.stored-descriptor-fields'
original = self._create([
DataField(0, 'id', AtomicType('INT')),
DataField(1, 'payload', AtomicType('BLOB')),
], {legacy_key: 'payload'})
- payload = b'legacy-blob-payload'
- messages = self._write(original, {'id': 1, 'payload': payload})
- assert any(f.file_name.endswith('.blob') for msg in messages for f in
msg.new_files)
+ source = self.root / 'blob-source'
+ payload = b'blob-payload'
+ source.write_bytes(payload)
+ messages = self._write(original, {'id': 1, 'payload': BlobDescriptor(
+ source.as_uri(), 0, len(payload)).serialize()})
+ assert all(f.file_name.endswith('.parquet') for msg in messages for f
in msg.new_files)
self.catalog.alter_table('test.t', [
SchemaChange.drop_column('payload'),
SchemaChange.add_column('reference', AtomicType('BYTES'),
'__BLOB_DESCRIPTOR_FIELD'),
@@ -160,10 +163,9 @@ class TestTimeTravelSchemaEvolution:
historical = current.copy({'scan.snapshot-id': '1'})
- # Restore the original key, without turning it into an
inline-descriptor layout switch.
assert historical.table_schema.options[legacy_key] == 'payload'
assert 'blob-descriptor-field' not in historical.table_schema.options
- assert not historical.options.blob_descriptor_fields()
+ assert historical.options.blob_descriptor_fields() == {'payload'}
assert self._read(historical) == [{'id': 1, 'payload': payload}]
def test_explicit_field_overrides_survive_repeated_copies(self):
@@ -185,3 +187,17 @@ class TestTimeTravelSchemaEvolution:
historical = current.copy({'scan.snapshot-id': '1'})
assert historical._applied_dynamic_options == {'scan.snapshot-id': '1'}
+
+ def test_descriptor_fallback_override_keeps_canonical_group(self):
+ original = self._create([
+ DataField(0, 'id', AtomicType('INT')), DataField(1, 'payload',
AtomicType('BLOB')),
+ ], {'blob.stored-descriptor-fields': 'payload'})
+ self._write(original, {'id': 1, 'payload': None})
+ self.catalog.alter_table('test.t', [SchemaChange.add_column(
+ 'reference', AtomicType('BYTES'), '__BLOB_DESCRIPTOR_FIELD')])
+ current = self.catalog.get_table('test.t')
+ historical = current.copy({'blob.stored-descriptor-fields':
None}).copy({'scan.snapshot-id': '1'})
+ repeated = historical.copy({'read.batch-size': '1'})
+ for table in (historical, repeated):
+ assert table.table_schema.options['blob-descriptor-field'] ==
'payload,reference'
+ assert 'blob.stored-descriptor-fields' not in
table.table_schema.options
diff --git
a/paimon-python/pypaimon/tests/table_update_by_row_id_chunked_test.py
b/paimon-python/pypaimon/tests/table_update_by_row_id_chunked_test.py
index 3ce93701ee..8f88e872d9 100644
--- a/paimon-python/pypaimon/tests/table_update_by_row_id_chunked_test.py
+++ b/paimon-python/pypaimon/tests/table_update_by_row_id_chunked_test.py
@@ -40,7 +40,7 @@ class TableUpdateByRowIdChunkedTest(unittest.TestCase):
table.fields = []
options = table.options
options.file_format.return_value = 'parquet'
- options.variant_shredding_enabled.return_value = False
+ options.variant_shredding_schema.return_value = None
options.data_evolution_row_sidecar_enabled.return_value = False
options.with_vector_format.return_value = False
options.changelog_producer.return_value = ChangelogProducer.NONE
@@ -59,7 +59,7 @@ class TableUpdateByRowIdChunkedTest(unittest.TestCase):
table.fields = [field]
options = table.options
options.file_format.return_value = 'parquet'
- options.variant_shredding_enabled.return_value = False
+ options.variant_shredding_schema.return_value = None
options.data_evolution_row_sidecar_enabled.return_value = False
options.with_vector_format.return_value = False
options.changelog_producer.return_value = ChangelogProducer.NONE
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 6c44aa00a6..1aee337e65 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -19,7 +19,8 @@
import pyarrow as pa
-from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
+from pypaimon.common.options.core_options import ChangelogProducer
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
from pypaimon.table.file_store_table import FileStoreTable
@@ -27,7 +28,7 @@ from pypaimon.write.native_commit import (
create_native_write_table, from_native_commit_messages,
)
from pypaimon.write.native_write import native_write_available,
_native_partition_types_supported
-from pypaimon.write.table_update_by_row_id import TableUpdateByRowId,
_supports_parquet_row_id_update
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
from pypaimon.write.row_utils import value_for_arrow
@@ -37,7 +38,12 @@ def _native_row_id_table(table):
or not table.options.native_write_enabled()
or not table.options.data_evolution_enabled()
or not table.options.row_tracking_enabled()
- or not _supports_parquet_row_id_update(table)
+ or table.is_primary_key_table
+ or table.options.file_format() != 'parquet'
+ or table.options.data_evolution_row_sidecar_enabled(False)
+ or table.options.with_vector_format()
+ or table.options.video_frame_fields()
+ or table.options.changelog_producer() != ChangelogProducer.NONE
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
or not native_write_available()):
return None
@@ -79,6 +85,8 @@ def _native_update_paths_supported(table):
def create_native_update(table, commit_user, columns):
"""Use the public core updater for direct and grouped row-ID updates."""
+ if not _native_update_columns_supported(table, columns):
+ return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
@@ -92,6 +100,11 @@ def create_native_update(table, commit_user, columns):
def create_native_update_by_row_id(table, commit_user, commit_identifier):
"""Create a core updater sharing one snapshot across incremental calls."""
+ # Its columns are selected on each later call. Until core supports raw
+ # Blob updates, retain one Python updater for the whole operation on such
+ # tables rather than switching after some files have already been staged.
+ if not _native_update_columns_supported(table, None):
+ return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
@@ -114,8 +127,20 @@ def _supported_upsert_key_type(data_type):
))
+def _native_update_columns_supported(table, columns):
+ """Java updatableBlobFields: descriptor and view fields are inline."""
+ if type(table) is not FileStoreTable:
+ return False
+ inline = table.options.blob_descriptor_fields() |
table.options.blob_view_fields()
+ names = set(table.field_names if columns is None else columns)
+ return not any(field.name in names and is_blob_file_field(field)
+ and field.name not in inline for field in table.fields)
+
+
def create_native_upsert(table, commit_user, data, keys, columns):
"""Prepare one core upsert from full Arrow rows or named row values."""
+ if not _native_update_columns_supported(table, columns):
+ return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
@@ -144,8 +169,10 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
return NativeTableUpsert(table, writer, keys, data)
-def create_native_predicate_update(table, commit_user, predicate):
+def create_native_predicate_update(table, commit_user, predicate,
columns=None):
"""Prepare a public core operation before any assignment can run."""
+ if not _native_update_columns_supported(table, columns):
+ return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
@@ -256,7 +283,8 @@ class NativeTableUpdateByRowId(TableUpdateByRowId):
return from_native_commit_messages(self.table, messages)
def _write_row_columns(self, data, column_names, blob_object_columns):
- # This adapter is selected only for plain Parquet tables without blobs.
+ # Row input normalization happens in the public Python API; physical
+ # field layouts and Blob references are handled by the core updater.
return self.update_columns(data, column_names)
diff --git a/paimon-python/pypaimon/write/row_utils.py
b/paimon-python/pypaimon/write/row_utils.py
index 7ee2e45c41..c70ea44a7c 100644
--- a/paimon-python/pypaimon/write/row_utils.py
+++ b/paimon-python/pypaimon/write/row_utils.py
@@ -21,7 +21,7 @@ from typing import Any, Dict, Iterable, List
import pyarrow as pa
from pypaimon.schema.data_types import DataField, PyarrowFieldParser,
is_blob_file_field
-from pypaimon.table.row.blob import Blob
+from pypaimon.table.row.blob import Blob, BlobView
from pypaimon.table.row.internal_row import InternalRow
from pypaimon.table.row.vector import Vector
@@ -62,6 +62,21 @@ def require_columns(
)
+def inline_blob_value(value: Any, descriptor: bool, view: bool) -> Any:
+ """Serialize row objects to inline references without opening payloads."""
+ if descriptor and isinstance(value, Blob):
+ try:
+ return value.to_descriptor().serialize()
+ except Exception as error:
+ raise ValueError(
+ "blob-descriptor-field row values must be serialized "
+ "BlobDescriptor bytes or a Blob with a descriptor."
+ ) from error
+ if view and isinstance(value, BlobView):
+ return value.view_struct.serialize()
+ return value
+
+
def value_for_arrow(value: Any, field: DataField) -> Any:
if isinstance(value, Vector):
return value.to_list()
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index e90b45195e..a2177c2efe 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -306,7 +306,7 @@ class TableUpdate:
from pypaimon.write.native_update import
create_native_predicate_update
try:
native = create_native_predicate_update(
- self.table, self.commit_user, predicate)
+ self.table, self.commit_user, predicate, list(assignments))
except Exception as error:
logger.debug('Native predicate update preparation failed: %s',
error)
else:
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index ad6b47334b..32de0a2a98 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -42,6 +42,7 @@ from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.file_store_write import FileStoreWrite
from pypaimon.write.row_id_file_index import RowIdFileIndex
from pypaimon.write.row_utils import (
+ inline_blob_value,
require_columns,
row_to_named_values,
value_for_arrow,
@@ -62,8 +63,7 @@ def _supports_parquet_row_id_update(table):
return (not table.is_primary_key_table
and options.file_format(CoreOptions.FILE_FORMAT_PARQUET)
== CoreOptions.FILE_FORMAT_PARQUET
- and not (options.variant_shredding_enabled()
- and options.variant_shredding_schema())
+ and not options.variant_shredding_schema()
and not options.data_evolution_row_sidecar_enabled(False)
and not options.with_vector_format()
and options.changelog_producer() == ChangelogProducer.NONE
@@ -337,6 +337,8 @@ class TableUpdateByRowId:
]
fields = [pa.field(SpecialFields.ROW_ID.name, pa.int64())]
blob_object_columns: Dict[str, List[Any]] = {}
+ descriptor_fields = self.table.options.blob_descriptor_fields()
+ view_fields = self.table.options.blob_view_fields()
for col_name in column_names:
if self._is_blob_column(col_name):
@@ -351,7 +353,9 @@ class TableUpdateByRowId:
arrays.append(
pa.array(
[
- value_for_arrow(values_by_name[col_name], table_field)
+ value_for_arrow(inline_blob_value(
+ values_by_name[col_name],
+ col_name in descriptor_fields, col_name in
view_fields), table_field)
for _, values_by_name in row_entries
],
type=arrow_field.type,
@@ -480,8 +484,11 @@ class TableUpdateByRowId:
# added after the original file was written, reading only those
# missing columns can otherwise produce a zero-row table instead of
# one null value per original row.
+ read_table = self.table
+ if wanted.intersection(self.table.options.blob_descriptor_fields() |
self.table.options.blob_view_fields()):
+ read_table = self.table.copy({'blob-as-descriptor': 'true',
'blob-view.resolve.enabled': 'false'})
table_read = TableRead(
- self.table,
+ read_table,
predicate=None,
read_type=read_fields + [SpecialFields.ROW_ID],
)
@@ -858,6 +865,8 @@ class TableUpdateByRowId:
return pa.array(pylist, type=target_type)
def _is_blob_column(self, column_name: str) -> bool:
+ if column_name in (self.table.options.blob_descriptor_fields() |
self.table.options.blob_view_fields()):
+ return False
for table_field in self.table.fields:
if table_field.name == column_name:
return is_blob_file_field(table_field)
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 731727fbc2..a00287caf7 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -99,8 +99,7 @@ class DataWriter(ABC):
)
# Variant shredding (static mode) — col_name → (obj_fields,
target_arrow_type)
self._variant_shredding: Dict[str, Tuple] = {}
- if self.file_format == CoreOptions.FILE_FORMAT_PARQUET \
- and self.options.variant_shredding_enabled():
+ if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
shredding_json = self.options.variant_shredding_schema()
if shredding_json:
from pypaimon.data.variant_shredding import (
diff --git a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
index dd0bb4096a..6ecfb23ad8 100644
--- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
+++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
@@ -29,11 +29,11 @@ from pypaimon.schema.data_types import (
is_blob_type,
)
from pypaimon.table.row.blob import (
- Blob,
BlobConsumer,
video_payload_descriptor,
)
from pypaimon.write.row_utils import (
+ inline_blob_value,
require_columns,
row_to_named_values,
row_values_to_arrow_table,
@@ -317,25 +317,8 @@ class DedicatedFormatWriter(CompositeDataWriter):
raise e
def _normal_row_value(self, field_name: str, value):
- if field_name in self.blob_descriptor_fields and value is not None:
- if isinstance(value, Blob):
- try:
- return value.to_descriptor().serialize()
- except Exception as e:
- raise ValueError(
- "blob-descriptor-field row values must be serialized "
- "BlobDescriptor bytes or a Blob with a descriptor."
- ) from e
- return value
-
- if field_name in self.blob_view_fields and value is not None:
- from pypaimon.table.row.blob import BlobView
-
- if isinstance(value, BlobView):
- return value.view_struct.serialize()
- return value
-
- return value
+ return inline_blob_value(value, field_name in
self.blob_descriptor_fields,
+ field_name in self.blob_view_fields)
def abort(self):
"""Abort all writers and clean up resources."""
