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 3357c6f956 [Python] Align selected write nullability with Java and
Native (#10666)
3357c6f956 is described below
commit 3357c6f956a4269997d0eb2a541c7353b2d411af
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 23:00:48 2026 +0800
[Python] Align selected write nullability with Java and Native (#10666)
---
.../tests/native_update_source_nullability_test.py | 263 +++++++++++++++++++++
.../write/retained_delete_nullability_test.py | 164 +++++++++++++
.../tests/write/row_kind_nullability_test.py | 17 +-
paimon-python/pypaimon/write/row_kind.py | 25 +-
.../pypaimon/write/table_update_by_row_id.py | 26 +-
paimon-python/pypaimon/write/table_write.py | 4 +
.../pypaimon/write/writer/key_value_data_writer.py | 4 +-
7 files changed, 490 insertions(+), 13 deletions(-)
diff --git
a/paimon-python/pypaimon/tests/native_update_source_nullability_test.py
b/paimon-python/pypaimon/tests/native_update_source_nullability_test.py
new file mode 100644
index 0000000000..e1f8756b6b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_update_source_nullability_test.py
@@ -0,0 +1,263 @@
+# 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.
+
+"""Source Arrow nullability must not validate values discarded by update
actions."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.data_evolution_merge_into import WhenMatched, source_col
+from pypaimon.table.row.generic_row import GenericRow
+
+
+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=[pytest.mark.native_plan,
_NATIVE])]
+_OPERATIONS = ['upsert_arrow', 'upsert_rows', 'merge']
+
+
+def _table(catalog, native, value_nullable=False):
+ schema = pa.schema([pa.field('id', pa.int32(), nullable=False),
+ pa.field('value', pa.int32(), nullable=value_nullable),
+ pa.field('keep', pa.int32(), nullable=False)])
+ catalog.create_table('default.updates', Schema.from_pyarrow_schema(schema,
options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'write.native.enabled': str(native).lower(), 'commit.native.enabled':
'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.updates')
+ 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, value=10,
keep=100)], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, schema
+
+
+def _source(schema, operation, invalid):
+ if invalid:
+ rows = [dict(id=1, value=None, keep=100)]
+ elif operation == 'merge':
+ # The second row does not match and there is no INSERT clause. Its
+ # NULL cannot be validated as though it were written to the target.
+ rows = [dict(id=1, value=20, keep=None), dict(id=2, value=None,
keep=None)]
+ else:
+ # Last-write-wins discards the first invalid value. The selected row
+ # updates only value, so keep is never a target assignment.
+ rows = [dict(id=1, value=None, keep=None), dict(id=1, value=20,
keep=None)]
+ if operation == 'upsert_rows':
+ return rows
+ source = pa.Table.from_pylist([dict(id=0, value=0, keep=0)] + rows,
schema=schema).slice(1)
+ source.validate(full=True)
+ return source.replace_schema_metadata({'origin': 'external-update'})
+
+
+def _messages(table, source, native, stream, operation):
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ kwargs = dict(commit_identifier=31) if stream else {}
+ with ExitStack() as stack:
+ if native:
+ for method in ('_upsert_partition', '_upsert_row_partition'):
+
stack.enter_context(patch('pypaimon.write.table_upsert_by_key.TableUpsertByKey.'
+ method,
+ side_effect=AssertionError('Python
upsert fallback')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python MERGE
fallback')))
+ if operation == 'merge':
+ messages = update.merge_into(source, on=['id'], when_matched=[
+ WhenMatched.update({'value': source_col('value')})], **kwargs)
+ elif operation == 'upsert_rows':
+ rows = [GenericRow([row[name] for name in table.field_names],
table.fields) for row in source]
+ messages = update.with_update_type(['value']).upsert_by_key(rows,
['id'], **kwargs)
+ else:
+ messages =
update.with_update_type(['value']).upsert_by_arrow_with_key(source, ['id'],
**kwargs)
+ return builder, messages
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist()
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
+def test_update_validates_selected_target_values_not_source_transport(
+ native_rest_catalog, native, stream, operation):
+ table, schema = _table(native_rest_catalog, native)
+ source = _source(schema, operation, invalid=False)
+ builder, messages = _messages(table, source, native, stream, operation)
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 31) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ assert _rows(table) == [dict(id=1, value=20, keep=100)]
+ if isinstance(source, pa.Table):
+ assert source.schema.metadata == {b'origin': b'external-update'}
+ assert all(not field.nullable for field in source.schema)
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
+def test_selected_target_not_null_value_is_still_required(
+ native_rest_catalog, native, stream, operation):
+ table, schema = _table(native_rest_catalog, native)
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises(ValueError, match='(?i)null'):
+ _messages(table, _source(schema, operation, invalid=True), native,
stream, operation)
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ assert _rows(table) == [dict(id=1, value=10, keep=100)]
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('operation', _OPERATIONS)
+def test_selected_nullable_target_accepts_null(native_rest_catalog, native,
operation):
+ table, schema = _table(native_rest_catalog, native, value_nullable=True)
+ builder, messages = _messages(table, _source(schema, operation,
invalid=True), native, False, operation)
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert _rows(table) == [dict(id=1, value=None, keep=100)]
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('reverse', [False, True], ids=['null-last',
'null-first'])
+def test_merge_chunked_source_keeps_one_transport_schema(
+ native_rest_catalog, native, stream, reverse):
+ table, schema = _table(native_rest_catalog, native)
+ schema = schema.with_metadata({'origin': 'chunked-merge'})
+ # Each chunk needs a different relaxed transport field. The NULL key and
+ # unmatched value never reach a target assignment, regardless of chunk
order.
+ rows = [dict(id=1, value=20, keep=100), dict(id=2, value=None, keep=None),
+ dict(id=None, value=30, keep=100)]
+ chunks = [pa.RecordBatch.from_pylist([dict(id=0, value=0, keep=0), row],
schema=schema).slice(1)
+ for row in rows]
+ source = pa.Table.from_batches(chunks[::-1] if reverse else chunks)
+ encoded = pa.chunked_array([pa.DictionaryArray.from_arrays(
+ pa.array([0], type=pa.int8()), pa.array([None], type=pa.int32())) for
_ in chunks])
+ source = source.append_column(pa.field('encoded_unused', encoded.type,
nullable=False), encoded)
+ schema = source.schema
+ assert source.column('encoded_unused').null_count == 0
+ assert source.column('encoded_unused').to_pylist() == [None, None, None]
+ assert len(source.to_batches()) == 3
+ builder, messages = _messages(table, source, native, stream, 'merge')
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 31) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ assert _rows(table) == [dict(id=1, value=20, keep=100)]
+ assert source.schema.equals(schema, check_metadata=True)
+
+
+def _blob_table(catalog, native, payload_type, nullable, before):
+ schema = pa.schema([pa.field('id', pa.int32()),
+ pa.field('payload', payload_type, nullable=nullable)])
+ catalog.create_table('default.blobs', Schema.from_pyarrow_schema(schema,
options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'write.native.enabled': str(native).lower(), 'commit.native.enabled':
'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.blobs')
+ 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, payload=before)],
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, schema
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('input_', ['arrow', 'row'])
[email protected]('nullable', [False, True], ids=['not-null',
'nullable'])
+def test_selected_blob_nullability_applies_to_dedicated_files(
+ native_rest_catalog, native, input_, nullable):
+ table, schema = _blob_table(native_rest_catalog, native,
pa.large_binary(), nullable, b'before')
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch(
+
'pypaimon.write.table_update_by_row_id.TableUpdateByRowId._load_existing_files_info',
+ side_effect=AssertionError('Python row-ID update fallback')))
+ updater = builder.new_update().new_update_by_row_id()
+
+ def update():
+ if input_ == 'row':
+ return updater.update_row_columns(GenericRow([None],
[table.field_dict['payload']]),
+ [0], ['payload'])
+ source = pa.Table.from_arrays([pa.array([0], type=pa.int64()),
+ pa.array([None],
type=pa.large_binary())], schema=pa.schema([
+ pa.field('_ROW_ID', pa.int64()),
schema.field('payload')]))
+ return updater.update_columns(source, ['payload'])
+
+ if nullable:
+ messages = update()
+ else:
+ with pytest.raises(ValueError, match='(?i)null'):
+ update()
+ messages = []
+ if nullable:
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ else:
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ assert _rows(table) == [dict(id=1, payload=None if nullable else
b'before')]
+
+
[email protected]('native', _ENGINES, ids=['python', 'native'])
[email protected]('payload_type,before,after', [
+ (pa.list_(pa.large_binary()), [b'before'], [None]),
+ (pa.map_(pa.string(), pa.large_binary()), [('key', b'before')], [('key',
None)]),
+], ids=['array', 'map'])
+def test_not_null_blob_containers_keep_nullable_children(
+ native_rest_catalog, native, payload_type, before, after):
+ table, schema = _blob_table(native_rest_catalog, native, payload_type,
False, before)
+ builder = table.new_batch_write_builder()
+ source = pa.Table.from_arrays([pa.array([0], type=pa.int64()),
pa.array([after], type=payload_type)],
+ schema=pa.schema([pa.field('_ROW_ID',
pa.int64()), schema.field('payload')]))
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch(
+
'pypaimon.write.table_update_by_row_id.TableUpdateByRowId._load_existing_files_info',
+ side_effect=AssertionError('Python row-ID update fallback')))
+ messages =
builder.new_update().new_update_by_row_id().update_columns(source, ['payload'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert _rows(table) == [dict(id=1, payload=after)]
diff --git
a/paimon-python/pypaimon/tests/write/retained_delete_nullability_test.py
b/paimon-python/pypaimon/tests/write/retained_delete_nullability_test.py
new file mode 100644
index 0000000000..061cd82ec1
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/retained_delete_nullability_test.py
@@ -0,0 +1,164 @@
+# 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.
+
+"""Deduplicate tombstones persist nullable values without changing logical NOT
NULL."""
+
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.row_kind import RowKind
+from pypaimon.write.native_write import NativeTableWrite
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_NATIVE = [False, pytest.param(True, marks=[pytest.mark.native_plan,
pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed')])]
+
+
+def _table(tmp_path, native=False, **options):
+ schema = pa.schema([pa.field('id', pa.int32(), nullable=False),
+ pa.field('seq', pa.int32(), nullable=False),
+ pa.field('value', pa.string(), nullable=False),
+ pa.field('op', pa.string())])
+ config = {'bucket': '1', 'sequence.field': 'seq', 'ignore-delete': 'false',
+ 'rowkind.field': 'op',
+ 'write.native.enabled': str(native).lower(),
'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false'}
+ config.update(options)
+ config = {key: value for key, value in config.items() if value is not None}
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options=config), False)
+ return catalog.get_table('default.t'), schema
+
+
+def _write(writer, table, schema, rows, input_, bucket=None):
+ if input_ == 'row':
+ for values in rows:
+ row = GenericRow([values[field.name] for field in table.fields],
table.fields,
+ RowKind.from_string(values['op']))
+ writer.write_row(row)
+ elif input_ == 'pandas':
+ import pandas as pd
+ writer.write_pandas(pd.DataFrame.from_records(rows,
columns=schema.names))
+ else:
+ if input_ == 'large_string':
+ schema = pa.schema([field.with_type(pa.large_string()) if
pa.types.is_string(field.type)
+ else field for field in schema])
+ data = pa.Table.from_pylist(rows, schema=schema)
+ writer.write_arrow(data.to_batches()[0] if input_ == 'batch' else
data, bucket)
+
+
+def _read(table, predicate=None):
+ reader = table.new_read_builder()
+ if predicate is not None:
+ reader.with_filter(predicate)
+ return
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist()
+
+
[email protected]('native', _NATIVE)
[email protected]('input_', ['row', 'batch', 'table', 'large_string',
'pandas'])
[email protected]('stream', [False, True])
+def test_retained_delete_persists_only_required_values(tmp_path, native,
input_, stream):
+ table, schema = _table(tmp_path, native)
+ for identifier, (seq, value, kind, expected) in enumerate([
+ (10, 'before', '+I', [dict(id=1, seq=10, value='before',
op='+I')]),
+ (9, None, '-D', [dict(id=1, seq=10, value='before', op='+I')]),
+ (11, None, '-D', [])], 1):
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ assert isinstance(writer, NativeTableWrite) == native
+ _write(writer, table, schema, [dict(id=1, seq=seq, value=value,
op=kind)], input_)
+ messages = writer.prepare_commit(identifier) if stream else
writer.prepare_commit()
+ # Data and INPUT changelog writers share the relaxed value type.
+ for path in Path(tmp_path).rglob('*.parquet'):
+ physical = pq.read_schema(path)
+ assert physical.field('value').nullable
+ assert not physical.field('id').nullable
+ assert not physical.field('seq').nullable
+ commit.commit(messages, identifier) if stream else
commit.commit(messages)
+ assert _read(table) == expected
+ predicate =
table.new_read_builder().new_predicate_builder().equal('value', 'before')
+ assert _read(table, predicate) == expected
+ assert not table.table_schema.fields[2].type.nullable
+ if native:
+ assert writer._python_writer is None
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('native', _NATIVE)
[email protected]('input_', ['row', 'batch', 'large_string', 'pandas'])
[email protected]('kind,field', [
+ ('-D', 'id'), ('-D', 'seq'), ('-U', 'value'), ('+U', 'value'), ('+I',
'value')])
+def test_retained_events_keep_logical_not_null_checks(tmp_path, native, kind,
field, input_):
+ table, schema = _table(tmp_path, native)
+ writer = table.new_batch_write_builder().new_write()
+ row = dict(id=1, seq=1, value='ok', op=kind)
+ row[field] = None
+ try:
+ with pytest.raises((ValueError, RuntimeError), match='non-null
column|non-nullable'):
+ _write(writer, table, schema, [row], input_)
+ assert not list(Path(tmp_path).rglob('*.parquet'))
+ finally:
+ writer.abort()
+
+
[email protected]('native', _NATIVE)
+def test_internal_row_delete_preserves_input_changelog_nulls(tmp_path, native):
+ table, schema = _table(tmp_path, native, **{'rowkind.field': None,
'changelog-producer': 'input'})
+ for seq, value, kind in [(1, 'before', '+I'), (2, None, '-D')]:
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ _write(writer, table, schema, [dict(id=1, seq=seq, value=value,
op=kind)], 'row')
+ messages = writer.prepare_commit()
+ files = [file for message in messages for file in
message.changelog_files]
+ assert len(files) == 1
+ path = next(Path(tmp_path).rglob(files[0].file_name))
+ assert pq.read_table(path, columns=['_VALUE_KIND',
'value']).to_pydict() == {
+ '_VALUE_KIND': [RowKind.from_string(kind).value], 'value':
[value]}
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ assert _read(table) == []
+
+
[email protected]('native', _NATIVE)
[email protected]('engine', ['partial-update', 'aggregation',
'first-row'])
+def test_delete_null_exception_does_not_relax_other_write_modes(tmp_path,
native, engine):
+ options = {'merge-engine': engine}
+ if engine == 'first-row':
+ options['sequence.field'] = None
+ table, schema = _table(tmp_path, native, **options)
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ with pytest.raises((ValueError, RuntimeError), match=r'non-null
column\(value\)'):
+ _write(writer, table, schema, [dict(id=1, seq=1, value=None,
op='-D')], 'table')
+ assert not list(Path(tmp_path).rglob('*.parquet'))
+ finally:
+ writer.abort()
diff --git a/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
b/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
index 62652600f1..17596a33ae 100644
--- a/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
+++ b/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
@@ -187,11 +187,22 @@ def
test_pandas_non_filtering_writers_keep_not_null_validation(tmp_path, primary
writer = table.new_batch_write_builder().new_write()
facade = NativeTableWrite(table, 'native-user', None, False, Mock())
try:
- for candidate in [writer, facade]:
+ with pytest.raises(ValueError, match='non-null column|non-nullable'):
+ writer.write_pandas(data)
+ if primary_key:
+ # The native facade preserves nullable KV transport; the actual
+ # core writer enforces logical INSERT nullability, covered by the
+ # retained DELETE integration tests.
+ facade.write_pandas(data)
+ transported = facade._native_writer.write_arrow.call_args[0][0]
+ assert transported.column('value').to_pylist() == [None]
+ assert transported.schema.field('value').nullable
+ assert not facade._schema.field('value').nullable
+ else:
with pytest.raises(ValueError, match='non-nullable'):
- candidate.write_pandas(data)
+ facade.write_pandas(data)
+ facade._native_writer.write_arrow.assert_not_called()
assert not writer.file_store_write.data_writers
- facade._native_writer.write_arrow.assert_not_called()
finally:
writer.close()
diff --git a/paimon-python/pypaimon/write/row_kind.py
b/paimon-python/pypaimon/write/row_kind.py
index ad66bf0cbc..34a9b24c0e 100644
--- a/paimon-python/pypaimon/write/row_kind.py
+++ b/paimon-python/pypaimon/write/row_kind.py
@@ -56,22 +56,22 @@ def filter_write_batch(table, data):
return data
options = table.options
raw = options.options.to_map()
- if 'rowkind.field' not in raw:
- return data
kinds = row_kinds(options, data)
ignore_delete = options.ignore_delete()
ignore_before = str(raw.get('ignore-update-before', 'false')).lower() ==
'true'
- if not ignore_delete and not ignore_before:
- return data
# Java checks top-level NULLs before its RowKindFilter. A DELETE in a
# non-cross-partition deduplicate table only requires non-NULL keys and
# sequence fields; UPDATE_BEFORE retains all declared NOT NULL fields.
for name, check_delete in _not_null_fields(table):
+ if name not in data.schema.names:
+ continue
invalid = data.column(name).is_null()
if not check_delete:
invalid = pc.and_(invalid, pc.not_equal(kinds,
RowKind.DELETE.value))
if pc.any(invalid).as_py():
raise ValueError('Cannot write null to non-null column(%s)' % name)
+ if not ignore_delete and not ignore_before:
+ return data
ignored = [RowKind.UPDATE_BEFORE.value]
if ignore_delete:
ignored.append(RowKind.DELETE.value)
@@ -109,10 +109,9 @@ def skip_write_row(table, values, row_kind=RowKind.INSERT):
if row_kind_field_name is None
else RowKind.from_string(values[row_kind_field_name]).value
)
- if options.ignore_delete() or str(raw.get('ignore-update-before',
'false')).lower() == 'true':
- for name, check_delete in _not_null_fields(table):
- if (check_delete or kind != RowKind.DELETE.value) and values[name]
is None:
- raise ValueError('Cannot write null to non-null column(%s)' %
name)
+ for name, check_delete in _not_null_fields(table):
+ if (check_delete or kind != RowKind.DELETE.value) and values.get(name)
is None:
+ raise ValueError('Cannot write null to non-null column(%s)' % name)
return _is_filtered(kind, options.ignore_delete(),
str(raw.get('ignore-update-before', 'false')).lower()
== 'true')
@@ -127,5 +126,15 @@ def _not_null_fields(table):
for field in table.fields if not field.type.nullable]
+def write_value_schema(table, schema):
+ """Java's KV value type can persist DELETEs with only keys and sequence."""
+ if (not table.is_primary_key_table or table.cross_partition_update
+ or table.options.merge_engine() != MergeEngine.DEDUPLICATE):
+ return schema
+ required = set(table.table_schema.primary_keys +
table.options.sequence_field())
+ return pa.schema([field if field.name in required else
field.with_nullable(True)
+ for field in schema], metadata=schema.metadata)
+
+
def _is_filtered(kind, ignore_delete, ignore_before):
return (ignore_delete and kind in (1, 3)) or (ignore_before and kind == 1)
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 ff721625d9..f39be2270e 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -671,7 +671,25 @@ class TableUpdateByRowId:
# write types at the writer boundary, including untouched input
batches.
write_schema = PyarrowFieldParser.from_paimon_schema(
[self.table.field_dict[name] for name in data.schema.names])
- return normalize_arrow_write_nullability(data, write_schema)
+ normalized = normalize_arrow_write_nullability(data, write_schema)
+ # Reads relax field nullability for schema evolution. A projected
+ # update must still enforce the declared write type, as Java's
+ # TableWrite.withWriteType does for every selected output column.
+ for field, column in zip(write_schema, normalized.columns):
+ self._check_write_not_null(field.name, column.null_count > 0)
+ if any(actual.nullable != expected.nullable
+ for actual, expected in zip(normalized.schema, write_schema)):
+ fields = [actual.with_nullable(expected.nullable)
+ for actual, expected in zip(normalized.schema,
write_schema)]
+ schema = pa.schema(fields, metadata=normalized.schema.metadata)
+ if isinstance(normalized, pa.RecordBatch):
+ return pa.RecordBatch.from_arrays(normalized.columns,
schema=schema)
+ return pa.Table.from_arrays(normalized.columns, schema=schema)
+ return normalized
+
+ def _check_write_not_null(self, column_name, has_null):
+ if not self.table.field_dict[column_name].type.nullable and has_null:
+ raise ValueError('Cannot write null to non-null column(%s)' %
column_name)
@classmethod
def _merge_chunked_column(
@@ -929,6 +947,12 @@ class TableUpdateByRowId:
blob_columns_with_baseline,
)
+ # Dedicated Blob files bypass the ordinary Arrow writer. Validate
+ # the merged values that will be written, including row-based inputs,
+ # while placeholders and NULL children of non-NULL containers remain
valid.
+ for column_name, values in blob_columns.items():
+ self._check_write_not_null(column_name, any(value is None for
value in values))
+
partition_tuple = tuple(partition.values)
new_files = []
file_store_write = None
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 6c02c33a0b..5f6f35e2b4 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -40,6 +40,8 @@ def _pandas_write_schema(table, schema):
"""Keep ignored events available for the normal row-kind NULL checks."""
if table is None or not table.is_primary_key_table:
return schema
+ from pypaimon.write.row_kind import write_value_schema
+ schema = write_value_schema(table, schema)
options = table.options
raw = options.options.to_map()
if (raw.get('rowkind.field') is None
@@ -375,6 +377,8 @@ class TableWrite:
fields.append(pa.field(field.name, field.type,
nullable=declared.nullable,
metadata=declared.metadata))
schema = pa.schema(fields, metadata=self.table_pyarrow_schema.metadata)
+ from pypaimon.write.row_kind import write_value_schema
+ schema = write_value_schema(self.table, schema)
if data.schema.equals(schema, check_metadata=True):
return data
if isinstance(data, pa.RecordBatch):
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index f513e148d6..63cddacaa6 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -298,6 +298,8 @@ class KeyValueDataWriter(DataWriter):
def _add_system_fields(self, data: pa.RecordBatch) -> pa.RecordBatch:
"""Add system fields: _KEY_{pk_key}, _SEQUENCE_NUMBER, _VALUE_KIND."""
num_rows = data.num_rows
+ from pypaimon.write.row_kind import write_value_schema
+ value_schema = write_value_schema(self.table, data.schema)
new_arrays = []
new_fields = []
@@ -325,7 +327,7 @@ class KeyValueDataWriter(DataWriter):
if data.schema.field(i).name == SpecialFields.VALUE_KIND.name:
continue
new_arrays.append(data.column(i))
- new_fields.append(data.schema.field(i))
+ new_fields.append(value_schema.field(i))
return pa.RecordBatch.from_arrays(new_arrays,
schema=pa.schema(new_fields))
