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 f7969ecbdd [python] Expand native sources and nullable-key
interoperability (#10640)
f7969ecbdd is described below
commit f7969ecbdd18b1b8927b66ae02b4d3a35ecfc0fa
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 16:02:13 2026 +0800
[python] Expand native sources and nullable-key interoperability (#10640)
---
paimon-python/README.md | 18 ++
paimon-python/pypaimon/index/dynamic_bucket.py | 3 +-
paimon-python/pypaimon/read/native_plan.py | 16 +-
paimon-python/pypaimon/read/stream_read_builder.py | 4 +-
.../pypaimon/tests/native_full_startup_test.py | 320 +++++++++++++++++++++
.../tests/native_nullable_primary_key_test.py | 250 ++++++++++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 12 +
.../pypaimon/tests/native_query_auth_test.py | 20 ++
paimon-python/pypaimon/write/file_store_write.py | 5 +-
paimon-python/pypaimon/write/native_merge_into.py | 24 +-
10 files changed, 661 insertions(+), 11 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index b911b37322..6ec3a01c21 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -177,6 +177,12 @@ propagate without retrying that batch through Python.
For primary-key Row writes, `rowkind.field` takes precedence over the object's
`RowKind`; otherwise the object event is preserved. Plain Arrow input defaults
to INSERT. Both Python and native writers apply `ignore-delete` and
`ignore-update-before` before routing object rows.
+Set `primary-key.nullable=true` when creating a primary-key table to allow NULL
+keys, including keys declared NOT NULL in the input schema. Both writers merge
+NULL keys using null-safe equality and restore their dynamic bucket assignments
+across restarts. SQL `MERGE ON` retains ordinary equality: a NULL in any
join-key
+column does not match another NULL.
+
Batch and stream `merge_into` also use Rust core for eligible data-evolution
Parquet tables when `write.native.enabled=true`. Existing `WhenMatched` and
`WhenNotMatched` clauses accept Arrow/pandas input, Paimon table sources, or
@@ -199,6 +205,13 @@ requiring catalog resolution and packed-video inserts use
the Python path
selected before native execution starts. Native failures propagate without
Python retry or deleting files from earlier prepared actions.
+A source object's captured snapshot is preserved, including a fixed empty
+view or retained snapshot metadata. Current main-branch REST table sources
+with `query-auth.enabled=true` also use core when the catalog returns no row
+filter or column mask. Core requests authorization for every source column,
+even for a key-only DELETE, and refreshes it for each MERGE. A restricted or
+denied source fails before target actions, as with Python's `plan_for_write`.
+
Python and native writers use Java's escaped partition directories for data,
changelog, Blob and row sidecar files, including external locations. Both
planners, file-range metadata, commit callbacks and explicit aborts use the
same
@@ -419,6 +432,11 @@ reusable `InternalRow` view with its row kind, BLOB and
VECTOR accessors. It
visits splits serially and shares one row limit across them. Close the iterator
when stopping before exhaustion to release its current Native reader.
+Batch reads support `scan.mode=latest` and `latest-full` for the latest
complete
+snapshot, and `from-snapshot-full` with `scan.snapshot-id` for a selected
complete
+snapshot. The same modes apply when a Paimon table is the source of
`merge_into`.
+These batch semantics follow Java; streaming startup policies are separate.
+
Native planning and reading require Rust's resolved-schema bindings:
`Table.from_resolved_schema()` for filesystem, JDBC and path-based tables, and
`Table.copy_with_resolved_schema()` for REST tables. The adapter passes the
Python
diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py
b/paimon-python/pypaimon/index/dynamic_bucket.py
index 89de907ea0..cec159988b 100644
--- a/paimon-python/pypaimon/index/dynamic_bucket.py
+++ b/paimon-python/pypaimon/index/dynamic_bucket.py
@@ -385,7 +385,8 @@ class HashBucketAssigner:
if scan_table.partition_keys:
predicate_builder = read_builder.new_predicate_builder()
predicates = [
- predicate_builder.equal(key, value)
+ predicate_builder.is_null(key) if value is None
+ else predicate_builder.equal(key, value)
for key, value in zip(scan_table.partition_keys, partition)
]
read_builder.with_filter(
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 4c30c5e525..0f7602003a 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -230,6 +230,11 @@ def _resolved_schema_json(table) -> str:
from pypaimon.common.json_util import JSON
options = {str(key): _option_value_to_string(value)
for key, value in table.table_schema.options.items() if value
is not None}
+ if 'query-auth.enabled' in options:
+ # Preserve Python's effective policy. Forwarding an enabled spelling
+ # such as "on" unchanged would disable authorization in Rust.
+ options['query-auth.enabled'] =
_option_value_to_string(OptionsUtils.convert_to_boolean(
+ table.table_schema.options['query-auth.enabled']))
return JSON.to_json(table.table_schema.copy(new_options=options))
@@ -370,13 +375,18 @@ def _native_table(table):
branch = getattr(rt, 'branch', None)
if not callable(branch) or branch() != table.current_branch():
raise RuntimeError("Native table did not resolve the requested
branch")
+ return _with_native_read_snapshot(table, rt)
+
+
+def _with_native_read_snapshot(table, native_table):
+ """Attach a captured read view, including a fixed empty snapshot."""
# Only explicitly stored state captures a view; None means a fixed empty
view.
table_state = vars(table)
if '_read_snapshot' in table_state:
from pypaimon.common.json_util import JSON
snapshot = table_state['_read_snapshot']
- rt = rt.copy_with_pinned_snapshot(JSON.to_json(snapshot) if snapshot
is not None else None)
- return rt
+ native_table =
native_table.copy_with_pinned_snapshot(JSON.to_json(snapshot) if snapshot is
not None else None)
+ return native_table
def _native_read_builder(table):
@@ -523,7 +533,7 @@ def native_plan(
return _from_native_plan(table, rust_plan)
-def _native_stream_auth_supported(table):
+def _native_query_auth_supported(table):
"""Only reproduce authorization owned by the standard REST environment."""
from pypaimon.catalog.catalog_environment import CatalogEnvironment,
_TableQueryAuthFn
diff --git a/paimon-python/pypaimon/read/stream_read_builder.py
b/paimon-python/pypaimon/read/stream_read_builder.py
index 252cc37e77..b98215abf4 100644
--- a/paimon-python/pypaimon/read/stream_read_builder.py
+++ b/paimon-python/pypaimon/read/stream_read_builder.py
@@ -129,9 +129,9 @@ class StreamReadBuilder:
"""Create a stateful streaming scan using the configured backend."""
projection = self._projection_builder()
projection._validate_map_key_filter()
- from pypaimon.read.native_plan import _native_stream_auth_supported
+ from pypaimon.read.native_plan import _native_query_auth_supported
if (self.table.options.native_plan_enabled()
- and _native_stream_auth_supported(self.table)):
+ and _native_query_auth_supported(self.table)):
try:
return StreamTableScan(
self.table, predicate=self._predicate,
read_type=projection.read_type(),
diff --git a/paimon-python/pypaimon/tests/native_full_startup_test.py
b/paimon-python/pypaimon/tests/native_full_startup_test.py
new file mode 100644
index 0000000000..78d877437b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_full_startup_test.py
@@ -0,0 +1,320 @@
+# 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.
+
+"""Existing full batch startup modes must use Native planning and reading."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
+from pypaimon.tests import native_plan_rest_test
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = pytest.mark.native_plan
+_SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.string())])
+
+
+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))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _create(catalog, name, mode):
+ options = {'write.native.enabled': 'false', 'source.split.target-size':
'1b'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif mode == 'de':
+ options.update({'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true'})
+ catalog.create_table('default.' + name, Schema.from_pyarrow_schema(
+ _SCHEMA, primary_keys=['id'] if mode == 'pk' else [],
options=options), False)
+ return catalog.get_table('default.' + name)
+
+
[email protected](params=['append', 'pk', 'de'])
+def source(rest_catalog, request):
+ catalog, _ = rest_catalog
+ table = _create(catalog, 'source', request.param)
+ _write(table, [{'id': 1, 'value': 'first'}])
+ first = table.snapshot_manager().get_latest_snapshot().id
+ _write(table, [{'id': 2, 'value': 'second'}])
+ return catalog, table, first
+
+
+def _view(table, mode, first, native):
+ options = {'scan.mode': mode, 'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()}
+ if mode == 'from-snapshot-full':
+ options['scan.snapshot-id'] = str(first)
+ return table.copy(options)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('mode', ['latest', 'latest-full',
'from-snapshot-full'])
[email protected]('method', ['arrow', 'batch', 'iterator'])
[email protected]('filtered', [False, True])
+def test_full_batch_startup_reads(source, native, mode, method, filtered):
+ _, original, first = source
+ table = _view(original, mode, first, native)
+ builder = table.new_read_builder()
+ if filtered:
+ predicate = builder.new_predicate_builder().equal('id', 2)
+ builder.with_projection(['value']).with_limit(1)
+ builder.with_filter(predicate)
+ scan = builder.new_scan()
+ read = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('Python scan')))
+ stack.enter_context(patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read')))
+ plan = scan.plan()
+ if method == 'arrow':
+ rows = read.to_arrow(plan.splits()).to_pylist()
+ elif method == 'batch':
+ with read.to_arrow_batch_reader(plan.splits()) as reader:
+ rows = reader.read_all().to_pylist()
+ else:
+ names = ['value'] if filtered else ['id', 'value']
+ rows = [dict(zip(names, (row.get_field(index) for index in
range(len(row)))))
+ for row in read.to_iterator(plan.splits())]
+ historical = mode == 'from-snapshot-full'
+ assert plan.snapshot_id == (first if historical else first + 1)
+ expected = ([{'value': 'second'}] if not historical else []) if filtered
else (
+ [{'id': 1, 'value': 'first'}] + ([] if historical else [{'id': 2,
'value': 'second'}]))
+ assert sorted(rows, key=lambda row: row.get('id', 0)) == expected
+ assert table.options.scan_mode() == mode
+ assert original.snapshot_manager().get_latest_snapshot().id == first + 1
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('mode', ['latest', 'latest-full',
'from-snapshot-full'])
[email protected]('stream', [False, True], ids=['batch-commit',
'stream-commit'])
+def test_merge_full_startup_table_source(source, native, mode, stream):
+ catalog, original, first = source
+ target = _create(catalog, 'target', 'de')
+ _write(target, [{'id': 1, 'value': 'old'}])
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ selected = _view(original, mode, first, native)
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python
merge')))
+ messages = builder.new_update().merge_into(
+ selected, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')], **({'commit_identifier':
37} if stream else {}))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 37) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ reader = target.new_read_builder()
+ actual =
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist()
+ expected = [{'id': 1, 'value': 'first'}]
+ if mode != 'from-snapshot-full':
+ expected.append({'id': 2, 'value': 'second'})
+ assert sorted(actual, key=lambda row: row['id']) == expected
+ assert original.snapshot_manager().get_latest_snapshot().id == first + 1
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('mode', ['latest', 'latest-full'])
+def test_full_startup_empty_table(rest_catalog, native, mode):
+ table = _create(rest_catalog[0], 'empty', 'append')
+ table = _view(table, mode, 1, native)
+ builder = table.new_read_builder()
+ scan, read = builder.new_scan(), builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('Python scan')))
+ stack.enter_context(patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read')))
+ plan = scan.plan()
+ assert plan.snapshot_id is None
+ assert plan.splits() == []
+ assert read.to_arrow(plan.splits()).num_rows == 0
+ assert list(read.to_iterator(plan.splits())) == []
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('view', ['empty', 'snapshot', 'retained-snapshot',
'latest'])
+def test_merge_keeps_captured_source_view(source, native, view):
+ catalog, original, first = source
+ snapshot = original.snapshot_manager().get_snapshot_by_id(first)
+ selected = (original._copy_with_snapshot(None if view == 'empty' else
snapshot)
+ if view != 'latest' else original)
+ selected = selected.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ if view == 'retained-snapshot':
+
original.file_io.delete(original.snapshot_manager().get_snapshot_path(first))
+ target = _create(catalog, 'target', 'de')
+ _write(target, [{'id': 1, 'value': 'old'}])
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ builder = target.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python
merge')))
+ messages = builder.new_update().merge_into(selected, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')})],
when_not_matched=[WhenNotMatched('*')])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ reader = target.new_read_builder()
+ rows =
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist()
+ expected = [{'id': 1, 'value': 'old' if view == 'empty' else 'first'}]
+ if view == 'latest':
+ expected.append({'id': 2, 'value': 'second'})
+ assert sorted(rows, key=lambda row: row['id']) == expected
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('empty', [False, True])
+def test_merge_unrestricted_authorized_source(source, rest_catalog, native,
empty):
+ from pypaimon.api.api_response import AuthTableQueryResponse
+ from pypaimon.schema.schema_change import SchemaChange
+
+ catalog, original, _ = source
+ if empty:
+ original = _create(catalog, 'empty_source', 'append')
+ catalog.alter_table(original.identifier,
[SchemaChange.set_option('query-auth.enabled', 'true')])
+ original = catalog.get_table(original.identifier)
+ selections = []
+
+ def authorize(select):
+ selections.append(select)
+ return AuthTableQueryResponse()
+
+
rest_catalog[1].table_query_auth_store[original.identifier.get_full_name()] =
authorize
+ target = _create(catalog, 'target', 'de')
+ _write(target, [{'id': 1, 'value': 'old'}])
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ builder = target.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+ messages = builder.new_update().merge_into(original, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')})],
when_not_matched=[WhenNotMatched('*')])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ reader = target.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).sort_by('id').to_pylist()
== (
+ [{'id': 1, 'value': 'old'}] if empty else
+ [{'id': 1, 'value': 'first'}, {'id': 2, 'value': 'second'}])
+ assert selections and all(select is None or set(select) ==
set(original.field_names) for select in selections)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('restriction', ['filter', 'mask', 'malformed',
'denied'])
[email protected]('empty', [False, True])
+def test_merge_authorized_source_failures_never_modify_target(rest_catalog,
native, restriction, empty):
+ import json
+ from pypaimon.api.api_response import AuthTableQueryResponse, ErrorResponse
+ from pypaimon.schema.schema_change import SchemaChange
+
+ catalog, server = rest_catalog
+ source = _create(catalog, 'source', 'append')
+ if not empty:
+ _write(source, [{'id': 1, 'value': 'source'}])
+ catalog.alter_table(source.identifier,
[SchemaChange.set_option('query-auth.enabled', 'true')])
+ source = catalog.get_table(source.identifier)
+ target = _create(catalog, 'target', 'de')
+ _write(target, [{'id': 1, 'value': 'old'}])
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ before = target.snapshot_manager().get_latest_snapshot().id
+ if restriction == 'mask':
+ response = AuthTableQueryResponse(column_masking={'value': json.dumps({
+ 'name': 'UPPER', 'inputs': [{'index': 1, 'name': 'value', 'type':
'STRING'}]})})
+ elif restriction == 'filter':
+ response = AuthTableQueryResponse(filter=[json.dumps({
+ 'kind': 'LEAF', 'transform': {
+ 'name': 'FIELD_REF', 'fieldRef': {'index': 0, 'name': 'id',
'type': 'INT'}},
+ 'function': 'GREATER_THAN', 'literals': [1]})])
+ elif restriction == 'malformed':
+ response = AuthTableQueryResponse(filter=['not-json'])
+ else:
+ response = AuthTableQueryResponse()
+ server.table_query_auth_store[source.identifier.get_full_name()] = response
+ builder = target.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+ if restriction == 'denied':
+ denied = server._mock_response(ErrorResponse('TABLE', 'source',
'permission denied', 403), 403)
+ stack.enter_context(patch.object(server,
'_table_query_auth_handle', return_value=denied))
+ with pytest.raises(Exception) as caught:
+ # No value column is used by this MERGE, but a mask on that column
+ # still prevents plan_for_write from exposing the source table.
+ builder.new_update().merge_into(source, on=['id'],
when_matched=[WhenMatched.delete()])
+ assert not isinstance(caught.value, AssertionError)
+ assert target.snapshot_manager().get_latest_snapshot().id == before
+ reader = target.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() == [
+ {'id': 1, 'value': 'old'}]
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('enabled', ['on', 'yes', '1', ' true '])
+def test_merge_source_preserves_enabled_authorization(rest_catalog, native,
enabled):
+ import json
+ from pypaimon.api.api_response import AuthTableQueryResponse
+ from pypaimon.schema.schema_change import SchemaChange
+
+ catalog, server = rest_catalog
+ source = _create(catalog, 'source', 'append')
+ _write(source, [{'id': 1, 'value': 'restricted'}])
+ catalog.alter_table(source.identifier,
[SchemaChange.set_option('query-auth.enabled', enabled)])
+ source = catalog.get_table(source.identifier)
+ assert source.options.query_auth_enabled
+ server.table_query_auth_store[source.identifier.get_full_name()] =
AuthTableQueryResponse(filter=[json.dumps({
+ 'kind': 'LEAF', 'transform': {
+ 'name': 'FIELD_REF', 'fieldRef': {'index': 0, 'name': 'id',
'type': 'INT'}},
+ 'function': 'GREATER_THAN', 'literals': [1]})])
+ target = _create(catalog, 'target', 'de')
+ _write(target, [{'id': 1, 'value': 'old'}])
+ before = target.snapshot_manager().get_latest_snapshot().id
+ builder = target.copy({'write.native.enabled':
str(native).lower()}).new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+ with pytest.raises(Exception) as caught:
+ builder.new_update().merge_into(source, on=['id'],
when_matched=[WhenMatched.delete()])
+ assert not isinstance(caught.value, AssertionError)
+ assert target.snapshot_manager().get_latest_snapshot().id == before
+ reader = target.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() == [
+ {'id': 1, 'value': 'old'}]
diff --git a/paimon-python/pypaimon/tests/native_nullable_primary_key_test.py
b/paimon-python/pypaimon/tests/native_nullable_primary_key_test.py
new file mode 100644
index 0000000000..dec23092bc
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_nullable_primary_key_test.py
@@ -0,0 +1,250 @@
+# 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.
+
+"""Nullable primary keys retain Java equality across Python and Native I/O."""
+
+from contextlib import ExitStack
+from datetime import date, datetime
+from decimal import Decimal
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+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_write import NativeTableWrite
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = pytest.mark.native_plan
+_SCHEMA = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+
+
+def _table(catalog, name, bucket, engine='deduplicate', primary_key=True,
extra=None):
+ options = {'write.native.enabled': 'false', 'source.split.target-size':
'1b'}
+ if primary_key:
+ options.update({'bucket': str(bucket), 'primary-key.nullable': 'true',
'merge-engine': engine})
+ if engine == 'aggregation':
+ options['fields.value.aggregate-function'] = 'sum'
+ else:
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true',
+ 'deletion-vectors.enabled': 'true'})
+ options.update(extra or {})
+ # Current Java also overrides an explicitly declared NOT NULL key when
enabled.
+ schema = pa.schema([pa.field(field.name, field.type, nullable=not
primary_key or field.name == 'value')
+ for field in _SCHEMA])
+ catalog.create_table('default.' + name, Schema.from_pyarrow_schema(
+ schema, primary_keys=['pt', 'id'] if primary_key else [],
+ partition_keys=['pt'] if primary_key else [], options=options), False)
+ table = catalog.get_table('default.' + name)
+ assert all(field.type.nullable for field in table.fields)
+ return table
+
+
+def _write(table, rows, native, stream=False, identifier=1, row_input=False):
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite) == native
+ try:
+ if row_input:
+ for row in rows:
+ writer.write_row(GenericRow(list(row), table.fields))
+ else:
+ arrays = [pa.array([row[index] for row in rows], type=field.type)
+ for index, field in enumerate(_SCHEMA)]
+ writer.write_arrow(pa.Table.from_arrays(arrays, schema=_SCHEMA))
+ if native:
+ assert writer._python_writer is None
+ messages = writer.prepare_commit(identifier) if stream else
writer.prepare_commit()
+ commit.commit(messages, identifier) if stream else
commit.commit(messages)
+ return messages
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _read(table, native, null_field=None, all_files=False):
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder()
+ if null_field:
+
builder.with_filter(builder.new_predicate_builder().is_null(null_field))
+ scan, read = builder.new_scan(), builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ if not all_files:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('Python scan')))
+ stack.enter_context(patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read')))
+ splits = (scan.plan_for_write() if all_files else scan.plan()).splits()
+ rows = read.to_arrow(splits).to_pylist()
+ # All values in these fixtures are non-negative except the seeded MERGE
target.
+ return sorted((tuple(row[name] for name in _SCHEMA.names) for row in rows),
+ key=lambda row: tuple(-1000 if value is None else value for
value in row))
+
+
[email protected]('bucket', [1, 3, -1], ids=['one-bucket', 'fixed',
'dynamic'])
[email protected]('engine', ['deduplicate', 'partial-update',
'aggregation'])
[email protected]('native_first', [False, True], ids=['python-first',
'native-first'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
+def test_nullable_composite_keys_across_writer_restarts(rest_catalog, bucket,
engine, native_first, stream):
+ table = _table(rest_catalog[0], 'nullable_keys', bucket, engine)
+ _write(table, [(None, None, 10), (1, None, 100), (1, 0, 1)], native_first,
stream, 1)
+ first = table.snapshot_manager().get_latest_snapshot()
+ # Restart with the other writer: a null key must retain its dynamic bucket.
+ _write(table, [(None, None, 20), (1, None, 200), (None, 0, 2)], not
native_first, stream, 2)
+ values = (30, 300) if engine == 'aggregation' else (20, 200)
+ expected = [(None, None, values[0]), (None, 0, 2), (1, None, values[1]),
(1, 0, 1)]
+ for native in [False, True]:
+ assert _read(table, native) == expected
+ assert _read(table, native, 'id') == [expected[0], expected[2]]
+ assert _read(table, native, 'pt') == expected[:2]
+ assert _read(table._copy_with_snapshot(first), native) == [(None,
None, 10), (1, None, 100), (1, 0, 1)]
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('bucket', [3, -1], ids=['fixed', 'dynamic'])
+def test_nullable_key_row_input_matches_arrow_input(rest_catalog, native,
bucket):
+ table = _table(rest_catalog[0], 'nullable_row', bucket)
+ _write(table, [(None, None, 10), (None, 0, 20)], native, row_input=True)
+ _write(table, [(None, None, 30), (1, None, 40)], native)
+ for native_read in [False, True]:
+ assert _read(table, native_read) == [(None, None, 30), (None, 0, 20),
(1, None, 40)]
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('engine', ['deduplicate', 'partial-update',
'aggregation', 'first-row'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
+def test_nullable_primary_key_source_uses_sql_merge_equality(rest_catalog,
native, engine, stream):
+ source = _table(rest_catalog[0], 'nullable_source', -1, engine)
+ _write(source, [(None, None, 10), (None, 1, 11), (1, None, 12), (1, 1,
100)], False)
+ _write(source, [(None, None, 20), (None, 1, 21), (1, None, 22), (1, 1,
200)], True)
+ target = _table(rest_catalog[0], 'target', -1, primary_key=False)
+ _write(target, [(None, None, -1), (None, 1, -3), (1, None, -4), (1, 1,
-2)], False)
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python
merge')))
+ messages = builder.new_update().merge_into(
+ source, on=['pt', 'id'],
when_matched=[WhenMatched.update({'value': source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')], **({'commit_identifier':
37} if stream else {}))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 37) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ values = {'aggregation': (30, 32, 34, 300), 'first-row': (10, 11, 12,
100)}.get(engine, (20, 21, 22, 200))
+ for native_read in [False, True]:
+ # Storage merges null PKs, but SQL MERGE ON does not match NULL to
NULL.
+ assert _read(target, native_read) == [
+ (None, None, -1), (None, None, values[0]), (None, 1, -3), (None,
1, values[1]),
+ (1, None, -4), (1, None, values[2]), (1, 1, values[3])]
+ assert _read(source, True, all_files=engine == 'first-row') == [
+ (None, None, values[0]), (None, 1, values[1]), (1, None, values[2]),
(1, 1, values[3])]
+ assert source.snapshot_manager().get_latest_snapshot().id == 2
+
+
[email protected]('bucket', [3, -1], ids=['fixed', 'dynamic'])
[email protected]('key_type,zero', [
+ (pa.bool_(), False), (pa.int64(), 0), (pa.string(), ''), (pa.binary(),
b''),
+ (pa.decimal128(20, 2), Decimal('0.00')), (pa.date32(), date(1970, 1, 1)),
+ (pa.timestamp('us'), datetime(1970, 1, 1)),
+], ids=['boolean', 'bigint', 'string', 'bytes', 'decimal', 'date',
'timestamp'])
+def
test_nullable_scalar_key_distinguishes_null_from_default_value(rest_catalog,
bucket, key_type, zero):
+ catalog, _ = rest_catalog
+ schema = pa.schema([pa.field('id', key_type, nullable=False), ('value',
pa.int32())])
+ catalog.create_table('default.scalar_keys', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options={'bucket': str(bucket),
'primary-key.nullable': 'true'}), False)
+ table = catalog.get_table('default.scalar_keys')
+ input_schema = pa.schema([('id', key_type), ('value', pa.int32())])
+ # Three separate writers exercise native index serialization, Python index
+ # restore and native restore after a Python row write.
+ for native, values in [(True, [(None, 10), (zero, 100)]),
+ (False, [(None, 20), (zero, 200)]), (True, [(None,
30)])]:
+ builder = table.copy({'write.native.enabled':
str(native).lower()}).new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite) == native
+ try:
+ if native:
+ writer.write_arrow(pa.Table.from_arrays([
+ pa.array([row[index] for row in values], type=field.type)
+ for index, field in enumerate(input_schema)],
schema=input_schema))
+ assert writer._python_writer is None
+ else:
+ for row in values:
+ writer.write_row(GenericRow(list(row), table.fields))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ for native in [False, True]:
+ builder = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ scan, read = builder.new_scan(), builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('Python scan')))
+ stack.enter_context(patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read')))
+ rows = read.to_arrow(scan.plan().splits()).to_pylist()
+ assert sorted(rows, key=lambda row: row['id'] is not None) == [
+ {'id': None, 'value': 30}, {'id': zero, 'value': 200}]
+
+
[email protected]('first_native', [False, True], ids=['python-first',
'native-first'])
+def test_null_key_restores_nonzero_dynamic_bucket(rest_catalog, first_native):
+ table = _table(rest_catalog[0], 'nonzero_null_key', -1,
+ extra={'dynamic-bucket.target-row-num': '1'})
+ initial = _write(table, [(None, 0, 100), (None, None, 10)], first_native)
+ assert {message.bucket for message in initial} == {0, 1}
+ for native, value in [(not first_native, 20), (first_native, 30)]:
+ messages = _write(table, [(None, None, value)], native)
+ assert {message.bucket for message in messages} == {1}
+ for native in [False, True]:
+ assert _read(table, native) == [(None, None, 30), (None, 0, 100)]
+
+
+def test_null_partition_requires_complete_dynamic_bucket_index(rest_catalog):
+ from pypaimon.write.row_key_extractor import DynamicBucketRowKeyExtractor
+
+ table = _table(rest_catalog[0], 'missing_hash', -1)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ # Simulate existing data whose HASH metadata is incomplete. Restoring
+ # this partition must fail instead of assigning its keys new buckets.
+ writer.row_key_extractor =
DynamicBucketRowKeyExtractor(table.table_schema)
+ writer.write_arrow(pa.Table.from_arrays([
+ pa.array([None], type=pa.int32()), pa.array([None],
type=pa.int32()),
+ pa.array([10], type=pa.int32())], schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ with pytest.raises(RuntimeError, match='has data files but no complete
HASH index'):
+ writer.write_arrow(pa.Table.from_arrays([
+ pa.array([None], type=pa.int32()), pa.array([None],
type=pa.int32()),
+ pa.array([20], type=pa.int32())], schema=_SCHEMA))
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 5e1adde4ad..eddd587868 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -838,6 +838,18 @@ class NativePlanTest(unittest.TestCase):
})
self.assertEqual(table.table_schema.options, options)
+ def test_resolved_schema_preserves_effective_query_auth_flag(self):
+ for value, expected in [
+ ('true', 'true'), ('TRUE', 'true'), ('on', 'true'), ('yes',
'true'),
+ ('1', 'true'), (' true ', 'true'), (True, 'true'), (2, 'true'),
+ ('false', 'false'), ('off', 'false'), ('0', 'false'), (False,
'false')]:
+ with self.subTest(value=value):
+ options = {'query-auth.enabled': value}
+ table = SimpleNamespace(table_schema=TableSchema(0, [],
options=options))
+ resolved = json.loads(_resolved_schema_json(table))['options']
+ self.assertEqual(resolved, {'query-auth.enabled': expected})
+ self.assertEqual(table.table_schema.options, options)
+
def test_resolved_schema_preserves_timestamp_selectors(self):
for key, value in (('scan.timestamp', '2026-09-22T00:00:00'),
('scan.timestamp-millis', 1790035200000)):
diff --git a/paimon-python/pypaimon/tests/native_query_auth_test.py
b/paimon-python/pypaimon/tests/native_query_auth_test.py
index 74503ccda7..840de7d911 100644
--- a/paimon-python/pypaimon/tests/native_query_auth_test.py
+++ b/paimon-python/pypaimon/tests/native_query_auth_test.py
@@ -300,3 +300,23 @@ def
test_explain_uses_the_authorized_native_plan(guarded_table):
with
patch('pypaimon.read.scanner.file_scanner.FileScanner.scan_with_stats',
side_effect=AssertionError('Native explain fell back')):
assert
table.new_read_builder().with_projection(['value']).explain().native_planned
+
+
[email protected]('enabled', ['on', ' true '])
+def test_native_read_preserves_effective_authorization_flag(guarded_table,
rest_catalog, enabled):
+ table, server = guarded_table
+ catalog, _ = rest_catalog
+ catalog.alter_table(table.identifier,
[SchemaChange.set_option('query-auth.enabled', enabled)])
+ table = catalog.get_table(table.identifier)
+ calls = []
+
+ def authorize(select):
+ calls.append(select)
+ return AuthTableQueryResponse(filter=[json.dumps({
+ 'kind': 'LEAF', 'transform': {'name': 'FIELD_REF', 'fieldRef':
_field('id', 'INT')},
+ 'function': 'GREATER_THAN', 'literals': [2]})])
+
+ server.table_query_auth_store[table.identifier.get_full_name()] = authorize
+ assert _read(table, True).column('id').to_pylist() == [3, 4]
+ assert len(calls) == 1
+ assert table.table_schema.options['query-auth.enabled'] == enabled
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index a7bc72db8b..b2746a9363 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -395,7 +395,10 @@ class FileStoreWrite:
predicate_builder = read_builder.new_predicate_builder()
sub_predicates = []
for key, value in zip(self.table.partition_keys, partition):
- sub_predicates.append(predicate_builder.equal(key, value))
+ # Java PartitionPredicate uses IS NULL for null partition values.
+ # Equality to NULL cannot locate the files whose sequence we
restore.
+ sub_predicates.append(predicate_builder.is_null(key) if value is
None
+ else predicate_builder.equal(key, value))
partition_filter = predicate_builder.and_predicates(sub_predicates)
scan = read_builder.with_filter(partition_filter).new_scan()
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 331804fc6c..5a009bf245 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -59,7 +59,9 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
if not _native_blob_view_supported(source, source.field_names):
return None
# Core resolves a full snapshot or PyPaimon's existing timestamp
window.
- if (source.options.scan_mode() not in ('default', 'from-snapshot',
'from-timestamp', 'incremental')
+ if (source.options.scan_mode() not in (
+ 'default', 'latest', 'latest-full', 'from-snapshot',
+ 'from-snapshot-full', 'from-timestamp', 'incremental')
or any(str(key).startswith('incremental-') and key !=
'incremental-between-timestamp'
for key in source.table_schema.options)):
return None
@@ -69,9 +71,23 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
source_schema, source.partition_keys,
source.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)):
return None
- native_source = create_native_write_table(source)
- if native_source is None:
- return None
+ if source.options.query_auth_enabled:
+ from pypaimon.read.native_plan import
_native_query_auth_supported, _native_table
+ from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+ from pypaimon.write.native_commit import _rest_catalog_supported
+ if (not _rest_catalog_supported(source) or not
_native_query_auth_supported(source)
+ or source.current_branch() != 'main' or '_read_snapshot'
in vars(source)
+ or any(source.table_schema.options.get(key) is not None
for key in SCAN_KEYS)):
+ return None
+ # A source is a read, not a write. Core obtains a fresh REST grant
+ # and requires it to be unrestricted, as Python plan_for_write
does.
+ native_source = _native_table(source)
+ else:
+ native_source = create_native_write_table(source)
+ if native_source is None:
+ return None
+ from pypaimon.read.native_plan import _with_native_read_snapshot
+ native_source = _with_native_read_snapshot(source, native_source)
source, matched, not_matched, context = _prepare(
table, source, list(matched), list(not_matched), on,
source_schema=source_schema)
if any(callable(value) for clause in matched + not_matched for value in
clause.spec.values()):
