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 b21f36cc56 [Python] Complete native FUSE writes, truncate, postpone
planning and MERGE callbacks (#10642)
b21f36cc56 is described below
commit b21f36cc56ee6bd11f43b7a2d7f43695445cd446
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 08:56:17 2026 +0800
[Python] Complete native FUSE writes, truncate, postpone planning and MERGE
callbacks (#10642)
---
.../pypaimon/ray/data_evolution_merge_into.py | 9 +-
.../pypaimon/ray/data_evolution_merge_join.py | 3 +
.../pypaimon/ray/data_evolution_merge_transform.py | 26 +-
paimon-python/pypaimon/read/native_plan.py | 54 ++-
.../pypaimon/table/data_evolution_merge_into.py | 8 +
paimon-python/pypaimon/tests/native_commit_test.py | 6 +-
.../tests/native_read_path_completion_test.py | 70 ++-
paimon-python/pypaimon/tests/native_read_test.py | 1 +
.../tests/native_write_path_completion_test.py | 493 +++++++++++++++++++++
.../tests/primary_key_index_definitions_test.py | 14 +-
.../tests/ray_data_evolution_merge_into_test.py | 60 ++-
.../pypaimon/tests/write/table_write_test.py | 30 +-
paimon-python/pypaimon/write/native_blob_rows.py | 12 +
paimon-python/pypaimon/write/native_commit.py | 46 +-
paimon-python/pypaimon/write/native_merge_into.py | 14 +-
paimon-python/pypaimon/write/native_update.py | 2 +-
paimon-python/pypaimon/write/native_write.py | 16 +-
.../pypaimon/write/postpone_batch_table_write.py | 13 +-
paimon-python/pypaimon/write/postpone_bucket.py | 30 +-
paimon-python/pypaimon/write/ray_datasink.py | 3 -
paimon-python/pypaimon/write/table_commit.py | 31 +-
21 files changed, 792 insertions(+), 149 deletions(-)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index 6fb9e34b99..ae32cecc3f 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -213,7 +213,7 @@ def _prepare(
c.update,
settable_field_names,
on_map,
- allow_callables=is_self_merge,
+ allow_callables=True,
)
matched_specs.append(
_NormalizedClause(
@@ -254,6 +254,7 @@ def _prepare(
spec = _normalize_set_spec(
c.insert, settable_field_names, on_map,
allow_target_refs=False,
+ allow_callables=True,
)
for tk, sk in on_map.items():
if tk in settable_field_names and tk not in spec:
@@ -277,8 +278,6 @@ def _prepare(
if read_columns and not has_callable:
raise ValueError("read_columns requires a callable SET value.")
if has_callable:
- if not read_columns:
- raise ValueError("Callable SET values require read_columns.")
for col in read_columns:
if col not in full_target_field_names:
raise ValueError(
@@ -714,10 +713,14 @@ def _resolve_target_projection(
target_on: Sequence[str],
update_cols: Sequence[str],
target_field_names: Sequence[str],
+ callable_columns: Optional[Sequence[str]] = None,
) -> list:
needed = set(_needed_target_cols(
clauses, target_on, update_cols, target_field_names,
))
+ if any(callable(value) and not isinstance(value, type)
+ for clause in clauses for value in clause.spec.values()):
+ needed.update(target_field_names if callable_columns is None else
callable_columns)
if any(c.condition is not None for c in clauses):
from pypaimon.ray.merge_condition import extract_target_columns
target_set = set(target_field_names)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 6225e59e6f..fb56710fa9 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -113,6 +113,8 @@ def _resolve_source_projection(
for clause in clauses:
for value in clause.spec.values():
+ if callable(value) and not isinstance(value, type):
+ needed.update(source_field_names)
if isinstance(value, SourceColumnRef):
needed.add(value.column)
if clause.condition is not None:
@@ -269,6 +271,7 @@ def build_self_merge_update_plan(
row_id_name = SpecialFields.ROW_ID.name
needed_cols = set(resolve_target_projection(
clauses, [row_id_name], update_cols, target_field_names,
+ callable_columns=read_columns or None,
))
needed_cols.update(read_columns)
for clause in clauses:
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_transform.py
b/paimon-python/pypaimon/ray/data_evolution_merge_transform.py
index ad29368e51..adaf5fa3cc 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_transform.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_transform.py
@@ -53,6 +53,8 @@ def lit(value: Any) -> LiteralValue:
class WhenMatched:
+ """Matched UPDATE/DELETE clause; callable SET values consume t./s. Arrow
columns."""
+
def __init__(
self,
action: str,
@@ -83,6 +85,8 @@ class WhenMatched:
@dataclass
class WhenNotMatched:
+ """INSERT clause; callable values consume s. Arrow columns and return
Arrow arrays."""
+
insert: SetSpec
condition: Optional[str] = None
@@ -169,6 +173,20 @@ def build_delete_schema(row_id_name: str) -> pa.Schema:
return pa.schema([pa.field(row_id_name, pa.int64(), nullable=False)])
+def _merge_callable_input(batch: pa.Table, on_pairs):
+ # Arrow joins coalesce ON keys; callbacks still expose both SQL aliases.
+ for source, target in on_pairs:
+ source_name, target_name = "s." + source, "t." + target
+ if target_name in batch.column_names:
+ if source_name in batch.column_names:
+ batch =
batch.set_column(batch.schema.get_field_index(source_name),
+ source_name,
batch.column(target_name))
+ else:
+ batch = batch.append_column(source_name,
batch.column(target_name))
+ return batch.select(sorted(name for name in batch.column_names
+ if name.startswith(("t.", "s."))))
+
+
def _resolve_spec_array(
val: Any,
batch: pa.Table,
@@ -178,11 +196,9 @@ def _resolve_spec_array(
callable_input: Optional[pa.Table] = None,
):
if callable(val) and not isinstance(val, type):
- if callable_input is None:
- raise TypeError(
- "Callable SET values are only supported for self-merge."
- )
- result = val(callable_input)
+ # Public MERGE callbacks see the joined t./s. columns. The internal
+ # predicate-update adapter supplies its existing unqualified read view.
+ result = val(_merge_callable_input(batch, on_pairs) if callable_input
is None else callable_input)
if not isinstance(result, (pa.Array, pa.ChunkedArray)):
raise ValueError(
"Callable SET values must return a pyarrow.Array or "
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 062fe799ca..6fedad6302 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -218,12 +218,13 @@ def _predicate_to_native(predicate: Predicate) -> dict:
return result
-def _resolved_rest_table_response(table):
+def _resolved_rest_table_response(table, identifier=None):
"""Reuse REST metadata only when the standard loader can be reproduced."""
from pypaimon.catalog.catalog_environment import CatalogEnvironment
from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
environment = table.catalog_environment
+ identifier = table.identifier if identifier is None else identifier
response = getattr(environment, 'rest_table_response', None)
if (type(environment) is not CatalogEnvironment
or type(environment.catalog_loader) is not RESTCatalogLoader
@@ -231,9 +232,9 @@ def _resolved_rest_table_response(table):
return None
metadata = json.loads(response)
if (metadata.get('path') != table.table_path
- or metadata.get('name') != table.identifier.get_object_name()
+ or metadata.get('name') != identifier.get_object_name()
or ('database' in metadata
- and metadata['database'] !=
table.identifier.get_database_name())):
+ and metadata['database'] != identifier.get_database_name())):
return None
return response
@@ -296,8 +297,9 @@ def _rest_data_token(table):
def _validate_native_fuse_binding(table):
- """Catalog options must reconstruct the FileIO already bound to the
view."""
+ """Find the catalog table whose construction reproduces the bound
FileIO."""
from pypaimon.catalog.rest.fuse_support import FusePathResolver
+ from pypaimon.common.identifier import Identifier
from pypaimon.filesystem.caching_file_io import CachingFileIO
from pypaimon.filesystem.local_file_io import FuseLocalFileIO, LocalFileIO
@@ -312,13 +314,30 @@ def _validate_native_fuse_binding(table):
options = loader.context().options
if not options.get(FuseOptions.FUSE_ENABLED):
raise RuntimeError('Native catalog would disable the bound FUSE mount')
- expected = FusePathResolver(options,
None).resolve_local_path(table.table_path, table.identifier)
+ if file_io.path.rstrip('/') != table.table_path.rstrip('/'):
+ raise RuntimeError('The bound FUSE FileIO belongs to a different table
path')
+ candidates = []
+ response = table.catalog_environment.rest_table_response
+ if isinstance(response, str):
+ metadata = json.loads(response)
+ if (metadata.get('path') == table.table_path
+ and metadata.get('database',
table.identifier.get_database_name())
+ == table.identifier.get_database_name() and
isinstance(metadata.get('name'), str)):
+ original = Identifier.create(table.identifier.get_database_name(),
metadata['name'])
+ if original.get_table_name() == table.identifier.get_table_name():
+ candidates.append(original)
+ candidates.append(table.identifier)
+ # A manually resolved view can omit the original REST response. Its usual
+ # main-table binding can still be reconstructed by the same catalog.
+ candidates.append(Identifier.create(table.identifier.get_database_name(),
table.identifier.get_table_name()))
+ resolver = FusePathResolver(options, None)
local = LocalFileIO()
- if (file_io.path.rstrip('/') != table.table_path.rstrip('/')
- or local._to_file(file_io.fuse_path).absolute() !=
local._to_file(expected).absolute()):
- # In pvfs mode, copy(branch=...) retains Python's original FileIO;
- # loading the decorated identifier would bind a different directory.
- raise RuntimeError('Native catalog would resolve a different FUSE
mount')
+ mounted = local._to_file(file_io.fuse_path).absolute()
+ for identifier in candidates:
+ expected = resolver.resolve_local_path(table.table_path, identifier)
+ if mounted == local._to_file(expected).absolute():
+ return identifier
+ raise RuntimeError('Native catalog cannot reconstruct the bound FUSE
FileIO')
def _native_table(table):
@@ -341,18 +360,18 @@ def _native_table(table):
def _load_native_table(table):
- rest_response = _resolved_rest_table_response(table)
+ identifier = _validate_native_fuse_binding(table) or table.identifier
+ rest_response = _resolved_rest_table_response(table, identifier)
file_io_options = _resolved_schema_file_io_options(table)
if rest_response is not None:
cache = table.catalog_environment.__dict__.setdefault(
'_native_rest_table_cache', _NativeRestTableCache())
rt = cache.get(
rest_response,
- database=table.identifier.get_database_name(),
- table=table.identifier.get_object_name(),
+ database=identifier.get_database_name(),
+ table=identifier.get_object_name(),
options=_catalog_options(table),
token_loader=lambda: _rest_data_token(table))
- rt = _with_native_resolved_schema(table, rt)
elif file_io_options is not None:
from pypaimon_rust.datafusion import Table
rt = Table.from_resolved_schema(
@@ -368,9 +387,14 @@ def _load_native_table(table):
# REST may keep branch schemas in the catalog only. Load the base
# environment, then attach the schema/branch already resolved here.
rt = catalog.get_table((
- table.identifier.get_database_name(),
table.identifier.get_object_name()))
+ identifier.get_database_name(), identifier.get_object_name()))
if rt.location() != table.table_path:
raise RuntimeError('Native catalog resolved a different table
location')
+ if identifier != table.identifier:
+ # Construct the originally bound table, then use the existing copy API.
+ # Core retains its FileIO while resolving the branch's REST identity.
+ rt = rt.copy({'branch': table.current_branch()})
+ if file_io_options is None:
rt = _with_native_resolved_schema(table, rt)
if table.current_branch() != 'main':
if rt.branch() != table.current_branch():
diff --git a/paimon-python/pypaimon/table/data_evolution_merge_into.py
b/paimon-python/pypaimon/table/data_evolution_merge_into.py
index b12a766a49..d64294af60 100644
--- a/paimon-python/pypaimon/table/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/table/data_evolution_merge_into.py
@@ -91,6 +91,12 @@ def merge_into(
The returned messages must be committed by the caller's matching
``TableCommit``.
+
+ SET and INSERT mappings may contain callables accepting an Arrow table
+ with qualified ``t.`` target and ``s.`` source columns (INSERT has only
+ source columns). Each callable evaluates the rows selected by its clause
+ and returns an Arrow Array or ChunkedArray of the same length. Empty
+ clauses do not invoke callbacks; callback errors propagate to the caller.
"""
if target_table.options.native_write_enabled():
from pypaimon.write.native_merge_into import create_native_merge_into
@@ -182,6 +188,7 @@ def _prepare(
c.update,
settable_field_names,
on_map,
+ allow_callables=True,
)
matched_specs.append(
_NormalizedClause(
@@ -230,6 +237,7 @@ def _prepare(
settable_field_names,
on_map,
allow_target_refs=False,
+ allow_callables=True,
)
for tk, sk in on_map.items():
if tk in settable_field_names and tk not in spec:
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 804a00db08..ff369e8d65 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -478,17 +478,17 @@ def test_callbacks_added_after_construction_use_native(
callback.close.assert_called_once()
-def test_truncate_uses_python(tmp_path):
+def
test_truncate_keeps_python_when_native_publication_is_unavailable(tmp_path):
table = _table(tmp_path)
_seed(table)
- with patch('pypaimon.write.native_commit.create_native_commit') as create:
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=None) as create:
for method, args in [('truncate_partitions', ([{'pt': 'a'}],)),
('truncate_table', ())]:
commit = table.new_batch_write_builder().new_commit()
try:
getattr(commit, method)(*args)
finally:
commit.close()
- create.assert_not_called()
+ assert create.call_count == 2
assert _rows(table) == []
diff --git a/paimon-python/pypaimon/tests/native_read_path_completion_test.py
b/paimon-python/pypaimon/tests/native_read_path_completion_test.py
index b415380daa..851d1b7d0b 100644
--- a/paimon-python/pypaimon/tests/native_read_path_completion_test.py
+++ b/paimon-python/pypaimon/tests/native_read_path_completion_test.py
@@ -85,14 +85,76 @@ def
test_copied_branch_does_not_rebind_an_existing_fuse_mount(tmp_path):
assert branch.identifier.get_object_name() == 't$branch_dev'
module = ModuleType('pypaimon_rust.datafusion')
module.PaimonCatalog = Mock()
- native_table = module.PaimonCatalog.return_value.get_table.return_value
+ native_table = Mock(spec=['location', 'branch', 'copy'])
+ module.PaimonCatalog.return_value.get_table.return_value = native_table
native_table.location.return_value = location
native_table.branch.return_value = 'dev'
+ native_table.copy.return_value = native_table
+ with patch.dict('sys.modules', {'pypaimon_rust.datafusion': module}), \
+ patch('pypaimon.read.native_plan._with_native_resolved_schema',
return_value=native_table):
+ assert _native_table(branch) is native_table
+ native_table.copy.assert_called_once_with({'branch': 'dev'})
+
module.PaimonCatalog.return_value.get_table.assert_called_once_with(('default',
't'))
+
+
[email protected]('target_branch', ['main', 'audit'])
[email protected]('mode', ['pvfs', 'raw'])
+def test_copied_fuse_view_reconstructs_original_non_main_binding(tmp_path,
target_branch, mode):
+ from pypaimon.catalog.catalog_context import CatalogContext
+ from pypaimon.catalog.rest.fuse_support import FusePathResolver
+ from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.options.options import Options
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO
+ from pypaimon.read.native_plan import _native_table
+ from pypaimon.schema.table_schema import TableSchema
+ from pypaimon.table.file_store_table import FileStoreTable
+
+ identifier = Identifier.create('default', 't', branch='dev')
+ options = Options({'fuse.enabled': 'true', 'fuse.root': tmp_path.as_uri(),
+ 'fuse.mode': mode, 'fuse.validation-mode': 'none'})
+ loader = RESTCatalogLoader(CatalogContext.create_from_options(options))
+ location = 'oss://bucket/warehouse/default/t'
+ response = json.dumps({'database': 'default', 'name': 't$branch_dev',
+ 'path': location, 'id': 'original-uuid'})
+ environment = CatalogEnvironment(identifier=identifier,
catalog_loader=loader,
+ uuid='original-uuid',
rest_table_response=response)
+ mount = FusePathResolver(options, None).resolve_local_path(location,
identifier)
+ table = FileStoreTable(FuseLocalFileIO(location, mount), identifier,
location, TableSchema(), environment)
+ table = table.copy_without_time_travel({'branch': target_branch})
+ module = ModuleType('pypaimon_rust.datafusion')
+ module.Table = Mock()
+ native_table = Mock(spec=['copy', 'branch'])
+ native_table.copy.return_value = native_table
+ native_table.branch.return_value = target_branch
+ module.Table.from_rest_response.return_value = native_table
with patch.dict('sys.modules', {'pypaimon_rust.datafusion': module}), \
patch('pypaimon.read.native_plan._with_native_resolved_schema',
return_value=native_table):
- with pytest.raises(RuntimeError, match='different FUSE mount'):
- _native_table(branch)
- module.PaimonCatalog.assert_not_called()
+ assert _native_table(table) is native_table
+ assert module.Table.from_rest_response.call_args[1]['table'] ==
't$branch_dev'
+ native_table.copy.assert_called_once_with({'branch': target_branch})
+
+
+def
test_native_fuse_binding_rejects_unreconstructible_mount_before_catalog_io(tmp_path):
+ from pypaimon.catalog.catalog_context import CatalogContext
+ from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO
+ from pypaimon.read.native_plan import _native_table
+ from pypaimon.schema.table_schema import TableSchema
+ from pypaimon.table.file_store_table import FileStoreTable
+
+ identifier = Identifier.create('default', 't')
+ loader = RESTCatalogLoader(CatalogContext.create_from_options({
+ 'fuse.enabled': 'true', 'fuse.root': str(tmp_path / 'configured')}))
+ location = 'oss://bucket/warehouse/default/t'
+ environment = CatalogEnvironment(identifier=identifier,
catalog_loader=loader)
+ table = FileStoreTable(FuseLocalFileIO(location, str(tmp_path /
'custom')), identifier,
+ location, TableSchema(), environment)
+ with patch('pypaimon.read.native_plan._load_native_table') as load:
+ with pytest.raises(RuntimeError, match='cannot reconstruct'):
+ _native_table(table)
+ load.assert_not_called()
@pytest.mark.parametrize('mode,branch', [('pvfs', 'main'), ('pvfs', 'dev'),
('raw', 'dev')])
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index 41a908465f..17e4d06ebd 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -233,6 +233,7 @@ def test_native_read_propagates_native_fork_safety_error():
split = _Split()
split._native_split = object()
with patch.dict(sys.modules, {'pypaimon_rust': module}), patch(
+ 'pypaimon.read.native_plan.native_runtime_available',
return_value=True), patch(
'pypaimon.read.native_plan.native_read',
side_effect=ForkSafetyError('cannot reuse Jindo after fork')):
with pytest.raises(ForkSafetyError):
diff --git a/paimon-python/pypaimon/tests/native_write_path_completion_test.py
b/paimon-python/pypaimon/tests/native_write_path_completion_test.py
new file mode 100644
index 0000000000..939a7b3183
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_write_path_completion_test.py
@@ -0,0 +1,493 @@
+# 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 write/commit routing and failure ownership contracts."""
+
+import json
+from types import ModuleType
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.write.native_commit import _rest_catalog_supported
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+native = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.string())])
+
+
+def _table(tmp_path, catalog=None, options=None):
+ catalog = catalog or CatalogFactory.create({'warehouse': str(tmp_path /
'warehouse')})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['pt'], options=dict({
+ 'commit.native.enabled': 'true', 'write.native.enabled': 'false'
+ }, **(options or {}))), False)
+ return catalog.get_table('default.t')
+
+
+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 _rows(table):
+ builder = table.new_read_builder()
+ return
sorted(builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist(),
+ key=lambda row: row['id'])
+
+
[email protected]('operation,args', [
+ ('truncate_table', ()), ('truncate_partitions', ([{'pt': 'a'}],)),
+])
+def test_native_truncate_publication_error_never_retries_or_aborts(tmp_path,
operation, args):
+ table = _table(tmp_path)
+ commit = table.new_batch_write_builder().new_commit()
+ native_commit = Mock()
+ cause = RuntimeError('snapshot may have published before response failed')
+ getattr(native_commit, operation).side_effect = cause
+ try:
+ with patch.object(commit, '_prepare_native_commit',
return_value=(native_commit, [])), \
+ patch.object(commit.file_store_commit, 'truncate_table') as
truncate, \
+ patch.object(commit.file_store_commit, 'drop_partitions') as
drop, \
+ patch.object(commit.file_store_commit, 'abort') as abort:
+ with pytest.raises(RuntimeError) as error:
+ getattr(commit, operation)(*args)
+ assert error.value is cause
+ truncate.assert_not_called()
+ drop.assert_not_called()
+ abort.assert_not_called()
+ native_commit.abort.assert_not_called()
+ finally:
+ commit.close()
+
+
+def
test_abort_then_truncate_partitions_then_commit_keeps_batch_lifecycle(tmp_path):
+ table = _table(tmp_path)
+ commit = table.new_batch_write_builder().new_commit()
+ stream, batch = Mock(), Mock()
+ try:
+ with patch('pypaimon.write.native_commit.create_native_commit',
side_effect=[stream, batch]) as create, \
+
patch('pypaimon.write.native_commit.to_native_commit_messages',
return_value=[]), \
+
patch('pypaimon.write.native_commit.native_messages_supported',
return_value=True), \
+ patch.object(commit.file_store_commit, 'commit') as fallback:
+ commit.abort([])
+ commit.truncate_partitions([{'pt': 'a'}])
+ # Java truncatePartitions does not consume the one-shot batch
commit.
+ commit.table = table.copy({'snapshot.ignore-empty-commit':
'false'})
+ commit.commit([])
+ stream.abort.assert_called_once_with([])
+ stream.close.assert_called_once_with()
+ batch.truncate_partitions.assert_called_once_with([{'pt': 'a'}])
+ batch.commit.assert_called_once_with([])
+ assert create.call_args_list[1][1] == {'batch': True}
+ fallback.assert_not_called()
+ with pytest.raises(RuntimeError, match='one-time'):
+ commit.commit([])
+ finally:
+ commit.close()
+
+
+def _bind_fuse(table, root, cached=False):
+ from pypaimon.catalog.rest.fuse_support import FusePathResolver
+ from pypaimon.filesystem.caching_file_io import CachingFileIO
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO, LocalFileIO
+
+ options = table.catalog_environment.catalog_loader.context().options
+ options.data.update({'fuse.enabled': 'true', 'fuse.root': root.as_uri(),
+ 'fuse.mode': 'pvfs', 'fuse.validation-mode': 'none'})
+ mount = LocalFileIO()._to_file(FusePathResolver(options,
None).resolve_local_path(
+ table.table_path, table.identifier)).absolute()
+ mount.parent.mkdir(parents=True, exist_ok=True)
+ mount.symlink_to(LocalFileIO()._to_file(table.table_path),
target_is_directory=True)
+ io = FuseLocalFileIO(table.table_path, str(mount), options)
+ table.file_io = CachingFileIO(io, Mock()) if cached else io
+ return mount
+
+
[email protected]('cached', [False, True])
+def test_builtin_fuse_is_accepted_without_relaxing_custom_file_io(tmp_path,
native_rest_catalog, cached):
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO
+
+ table = _table(tmp_path, native_rest_catalog)
+ mount = _bind_fuse(table, tmp_path / 'fuse', cached)
+ assert _rest_catalog_supported(table)
+
+ class CustomFuse(FuseLocalFileIO):
+ pass
+
+ table.file_io = CustomFuse(table.table_path, str(mount))
+ assert not _rest_catalog_supported(table)
+
+
[email protected]('cached', [False, True])
[email protected]('invalid', [None, 'source-uuid', 'source-path',
'target-path', 'target-branch'])
+def test_copied_fuse_write_uses_existing_copy_and_checks_identity(
+ tmp_path, native_rest_catalog, cached, invalid):
+ from pypaimon.write.native_commit import create_native_write_table
+
+ original = _table(tmp_path, native_rest_catalog)
+ _bind_fuse(original, tmp_path / 'fuse', cached)
+ table = original.copy_without_time_travel({'branch': 'dev'})
+ assert table.catalog_environment.uuid is None
+ assert _rest_catalog_supported(table)
+
+ module = ModuleType('pypaimon_rust.datafusion')
+ module.Table = Mock()
+ module.PaimonCatalog = Mock()
+ source = Mock(spec=['location', 'rest_table_uuid', 'copy'])
+ target = Mock(spec=['location', 'branch', 'copy_with_resolved_schema'])
+ source.location.return_value = original.table_path if invalid !=
'source-path' else '/different'
+ source.rest_table_uuid.return_value = (
+ original.catalog_environment.uuid if invalid != 'source-uuid' else
'replacement-uuid')
+ source.copy.return_value = target
+ target.location.return_value = table.table_path if invalid !=
'target-path' else '/different'
+ target.branch.return_value = 'dev' if invalid != 'target-branch' else
'main'
+ target.copy_with_resolved_schema.return_value = target
+ module.PaimonCatalog.return_value.get_table.return_value = source
+ with patch.dict('sys.modules', {'pypaimon_rust.datafusion': module}):
+ result = create_native_write_table(table)
+
module.PaimonCatalog.return_value.get_table.assert_called_once_with(('default',
't'))
+ if invalid is None:
+ assert result is target
+ source.copy.assert_called_once_with({'branch': 'dev'})
+ target.copy_with_resolved_schema.assert_called_once()
+ schema = json.loads(target.copy_with_resolved_schema.call_args[0][0])
+ assert schema['options']['branch'] == 'dev'
+ else:
+ assert result is None
+ target.copy_with_resolved_schema.assert_not_called()
+ if invalid.startswith('source-'):
+ source.copy.assert_not_called()
+
+
[email protected]('identity', [None, '', 123])
+def test_copied_fuse_write_requires_original_rest_uuid(tmp_path,
native_rest_catalog, identity):
+ table = _table(tmp_path, native_rest_catalog)
+ _bind_fuse(table, tmp_path / 'fuse')
+ table = table.copy_without_time_travel({'branch': 'dev'})
+ response = json.loads(table.catalog_environment.rest_table_response)
+ response['id'] = identity
+ table.catalog_environment.rest_table_response = json.dumps(response)
+ assert not _rest_catalog_supported(table)
+
+
+@native
[email protected]('copied_branch', [False, True])
+def test_fuse_native_write_commit_and_truncate_preserve_bound_mount(
+ tmp_path, native_rest_catalog, copied_branch):
+ table = _table(tmp_path, native_rest_catalog, {'write.native.enabled':
'true'})
+ _bind_fuse(table, tmp_path / 'fuse')
+ if copied_branch:
+ table.branch_manager().create_branch('dev')
+ table = table.copy({'branch': 'dev'})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{'id': 1, 'pt': 'a'}, {'id':
2, 'pt': 'b'}], schema=_SCHEMA))
+ with patch.object(commit.file_store_commit, 'commit',
side_effect=AssertionError('Python commit')):
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}]
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ with patch.object(commit.file_store_commit, 'drop_partitions',
side_effect=AssertionError('Python truncate')):
+ commit.truncate_partitions([{'pt': 'a'}])
+ assert _rows(table) == [{'id': 2, 'pt': 'b'}]
+ with patch.object(commit.file_store_commit, 'truncate_table',
side_effect=AssertionError('Python truncate')):
+ commit.truncate_table()
+ assert _rows(table) == []
+ finally:
+ commit.close()
+
+
+def _merge_table(tmp_path, use_native):
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32()), ('payload',
pa.string())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'merge')})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(use_native).lower()}), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ {'id': 1, 'value': 10, 'payload': 'old'}, {'id': 2, 'value': 20,
'payload': 'keep'}
+ ], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table
+
+
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
[email protected]('source_key', ['id', 'source_id'])
[email protected]('dictionary_key', [False, True])
+def test_merge_callable_reads_target_source_and_coalesced_keys(tmp_path,
use_native, source_key, dictionary_key):
+ import pyarrow.compute as pc
+ import pypaimon.table.data_evolution_merge_into as merge_impl
+ from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched
+ table = _merge_table(tmp_path, use_native)
+ keys = pa.array([1, 3], pa.int32())
+ source = pa.table({source_key: keys.dictionary_encode() if dictionary_key
else keys,
+ 'delta': pa.array([4, 5], pa.int32()), 'label':
['unused', 'new']})
+ calls = []
+
+ def value(batch):
+ calls.append(batch.column_names)
+ assert batch['s.' + source_key].type == pa.int32()
+ return pc.add(pc.add(batch['t.value'], batch['s.delta']), batch['s.' +
source_key])
+
+ update = table.new_batch_write_builder().new_update()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ wraps=merge_impl._build_tables) as python_merge:
+ messages = update.merge_into(
+ source, on={'id': source_key},
+ when_matched=[WhenMatched.update({
+ 'value': value, 'payload': lambda batch:
pc.utf8_upper(batch['t.payload'])})],
+ when_not_matched=[WhenNotMatched({
+ 'value': lambda batch: pc.multiply(batch['s.delta'], 2),
+ 'payload': lambda batch: batch['s.label']})])
+ if use_native:
+ python_merge.assert_not_called()
+ table.new_batch_write_builder().new_commit().commit(messages)
+ assert _rows(table) == [{'id': 1, 'value': 15, 'payload': 'OLD'},
+ {'id': 2, 'value': 20, 'payload': 'keep'},
+ {'id': 3, 'value': 10, 'payload': 'new'}]
+ assert len(calls) == 1
+ assert calls[0] == sorted(['t._ROW_ID', 't.id', 't.value', 't.payload',
+ 's.' + source_key, 's.delta', 's.label'])
+
+
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
+def test_merge_callable_is_lazy_and_failure_preserves_committed_rows(tmp_path,
use_native):
+ from pypaimon.table.data_evolution_merge_into import WhenMatched
+ table = _merge_table(tmp_path, use_native)
+ cause = RuntimeError('callable failure')
+ calls = []
+
+ def fail(batch):
+ calls.append(batch.num_rows)
+ raise cause
+
+ update = table.new_batch_write_builder().new_update()
+ clause = [WhenMatched.update({'value': fail})]
+ assert update.merge_into(pa.table({'id': pa.array([99], pa.int32())}),
on=['id'],
+ when_matched=clause, when_not_matched=[]) == []
+ assert calls == []
+ with pytest.raises(RuntimeError) as error:
+ update.merge_into(pa.table({'id': pa.array([1], pa.int32())}),
on=['id'],
+ when_matched=clause, when_not_matched=[])
+ assert error.value is cause
+ assert calls == [1]
+ assert _rows(table) == [{'id': 1, 'value': 10, 'payload': 'old'},
+ {'id': 2, 'value': 20, 'payload': 'keep'}]
+
+
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
[email protected]('result_kind', ['chunked', 'wrong_length',
'not_arrow'])
+def test_merge_callable_validates_complete_clause_result(tmp_path, use_native,
result_kind):
+ from pypaimon.table.data_evolution_merge_into import WhenMatched
+ table = _merge_table(tmp_path, use_native)
+ source = pa.table({'id': pa.array([1, 2], pa.int32())})
+ calls = []
+
+ def values(batch):
+ calls.append(batch.num_rows)
+ if result_kind == 'chunked':
+ return pa.chunked_array([pa.array([101]), pa.array([202])])
+ if result_kind == 'wrong_length':
+ return pa.array([1])
+ return [101, 202]
+
+ update = table.new_batch_write_builder().new_update()
+ if result_kind == 'chunked':
+ messages = update.merge_into(source, on=['id'],
when_matched=[WhenMatched.update({'value': values})])
+ table.new_batch_write_builder().new_commit().commit(messages)
+ assert [row['value'] for row in _rows(table)] == [101, 202]
+ else:
+ with pytest.raises((ValueError, TypeError)):
+ update.merge_into(source, on=['id'],
when_matched=[WhenMatched.update({'value': values})])
+ assert [row['value'] for row in _rows(table)] == [10, 20]
+ assert calls == [2]
+
+
+def test_merge_callback_aliases_use_coalesced_key_type():
+ from pypaimon.ray.data_evolution_merge_transform import
_merge_callable_input
+ dictionary = pa.array([1, 2], pa.int32()).dictionary_encode()
+ batch = pa.table({'t.id': pa.array([1, 2], pa.int32()), 's.source_id':
dictionary,
+ 's.value': [10, 20], '_scratch': [0, 0]})
+ actual = _merge_callable_input(batch, [('source_id', 'id')])
+ assert actual.column_names == ['s.source_id', 's.value', 't.id']
+ assert actual['s.source_id'].equals(actual['t.id'])
+ # INSERT callbacks retain source layout because no target key exists.
+ unmatched = _merge_callable_input(batch.drop(['t.id']), [('source_id',
'id')])
+ assert unmatched['s.source_id'].type == dictionary.type
+
+
+def test_explicit_callable_read_columns_keep_self_merge_projection_narrow():
+ from pypaimon.ray.data_evolution_merge_into import
_resolve_target_projection
+ from pypaimon.ray.data_evolution_merge_transform import _NormalizedClause
+ clauses = [_NormalizedClause({'value': lambda rows: rows['value']})]
+ fields = ['id', 'value', 'large_blob']
+ assert _resolve_target_projection(clauses, ['_ROW_ID'], ['value'], fields,
+ callable_columns=['value']) == ['value']
+ assert _resolve_target_projection(clauses, ['id'], ['value'], fields) ==
fields
+
+
+def test_consumed_blob_rows_are_released_without_losing_unplanned_inputs():
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.table.row.blob import Blob, BlobDescriptor
+ from pypaimon.write.native_blob_rows import NativeBlobRows
+ rows = NativeBlobRows()
+ blob_type = AtomicType('BLOB')
+ known = BlobDescriptor.deserialize(rows.encode(Blob.from_data(b'known'),
blob_type)).uri
+ unknown =
BlobDescriptor.deserialize(rows.encode(Blob.from_data(b'unknown'),
blob_type)).uri
+ reader = rows.create(known)
+ assert rows.create(known) is reader
+ rows.release_consumed()
+ assert not rows._supports_uri(known)
+ assert rows._supports_uri(unknown)
+ with rows.create(unknown).new_input_stream(unknown) as stream:
+ assert stream.read() == b'unknown'
+ rows.clear()
+ assert not rows.readers
+ assert not rows._consumed
+
+
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
+def test_automatic_postpone_plan_counts_all_batches_per_partition(tmp_path,
use_native):
+ from pypaimon.write.native_write import NativePostponeFixedBucketTableWrite
+ schema = pa.schema([('id', pa.int64()), ('pt', pa.string())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'postpone')})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id', 'pt'], partition_keys=['pt'], options={
+ 'bucket': '-2', 'write.native.enabled': str(use_native).lower(),
+ 'postpone.target-row-num-per-bucket': '2',
+ 'postpone.batch-write-fixed-bucket.max-parallelism': '3'}), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_postpone_fixed_bucket_write_builder()
+ writer = builder.new_write()
+ if use_native:
+ assert isinstance(writer, NativePostponeFixedBucketTableWrite)
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{'id': 1, 'pt': 'a'}, {'id':
2, 'pt': 'b'}], schema=schema))
+ writer.write_arrow(pa.Table.from_pylist([{'id': n, 'pt': 'a'} for n in
[3, 4, 5, 6]], schema=schema))
+ messages = writer.prepare_commit()
+ assert {m.partition: m.total_buckets for m in messages} == {('a',): 3,
('b',): 1}
+ builder.new_commit().commit(messages)
+ finally:
+ writer.close()
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}] + [
+ {'id': n, 'pt': 'a'} for n in [3, 4, 5, 6]]
+
+
+@native
[email protected]('kind', ['boolean', 'date', 'time', 'timestamp',
'decimal', 'float',
+ 'unicode_int', 'decimal_overflow'])
+def test_native_truncate_partition_specs_use_java_typed_strings(tmp_path,
native_rest_catalog, kind):
+ from datetime import date, datetime, time
+ from decimal import Decimal
+ cases = {
+ 'boolean': (pa.bool_(), [True, False], 'TRUE'),
+ 'date': (pa.date32(), [date(2024, 1, 2), date(2024, 1, 3)],
'2024-01-02'),
+ 'time': (pa.time32('ms'), [time(12, 34, 56, 789000), time(12, 35)],
'12:34:56.789'),
+ 'timestamp': (pa.timestamp('us'), [datetime(2024, 1, 2, 12, 34),
datetime(2024, 1, 3)],
+ '2024-01-02 12:34:00'),
+ 'decimal': (pa.decimal128(6, 2), [Decimal('12.35'), Decimal('13.00')],
'12.345'),
+ 'float': (pa.float32(), [1.2, 2.0], '1.2'),
+ 'unicode_int': (pa.int32(), [12, 13], '１２'),
+ 'decimal_overflow': (pa.decimal128(3, 0), [None, Decimal('12')],
'1000'),
+ }
+ data_type, values, text = cases[kind]
+ schema = pa.schema([('id', pa.int64()), ('pt', data_type)])
+ native_rest_catalog.create_table('default.typed',
Schema.from_pyarrow_schema(
+ schema, partition_keys=['pt'], options={
+ 'write.native.enabled': 'false', 'commit.native.enabled':
'true'}), False)
+ table = native_rest_catalog.get_table('default.typed')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [1, 2], 'pt': values},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ with patch.object(commit.file_store_commit, 'drop_partitions',
side_effect=AssertionError('Python truncate')):
+ commit.truncate_partitions([{'pt': text}])
+ assert [row['id'] for row in _rows(table)] == [2]
+ finally:
+ commit.close()
+
+
+@native
+def
test_automatic_postpone_blob_rows_keep_unknown_sources_until_prepare(tmp_path):
+ from pypaimon.schema.data_types import AtomicType, DataField
+ from pypaimon.table.row.blob import Blob, BlobDescriptor
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.write.native_write import NativePostponeFixedBucketTableWrite
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path /
'blob-postpone')})
+ catalog.create_database('default', True)
+ fields = [DataField(0, 'id', AtomicType('INT')), DataField(1, 'pt',
AtomicType('STRING')),
+ DataField(2, 'payload', AtomicType('BLOB'))]
+ catalog.create_table('default.t', Schema(
+ fields, partition_keys=['pt'], primary_keys=['id', 'pt'], options={
+ 'bucket': '-2', 'write.native.enabled': 'true',
+ 'postpone.target-row-num-per-bucket': '1'}), False)
+ table = catalog.get_table('default.t')
+
+ def write(rows, buffered):
+ builder = table.new_postpone_fixed_bucket_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ assert isinstance(writer, NativePostponeFixedBucketTableWrite)
+ for identifier, partition in rows:
+ writer.write_row(GenericRow(
+ [identifier, partition,
Blob.from_data(str(identifier).encode())], table.fields))
+ assert len(writer._blob_rows.readers) == buffered
+ messages = writer.prepare_commit()
+ assert not writer._blob_rows.readers
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+ write([(1, 'known')], 1)
+ write([(2, 'known'), (3, 'new')], 1)
+ rows = _rows(table)
+ assert [row['id'] for row in rows] == [1, 2, 3]
+ assert [Blob.from_descriptor_bytes(row['payload'],
table.file_io).to_data() for row in rows] == [b'1', b'2', b'3']
+ assert all(not
BlobDescriptor.deserialize(row['payload']).uri.startswith('pypaimon-blob-row:')
for row in rows)
diff --git a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
index 7907651a71..eaf7b229e3 100644
--- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
+++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
@@ -111,15 +111,11 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
.with_query("content", "paimon")
.with_limit(2))
- # Exercise native qualification even without an installed binding. A
- # schema-only table cannot delegate, but still selects the PK readers.
- with mock.patch(
-
"pypaimon.table.source.native_search_plan.native_runtime_available",
- return_value=True):
- self.assertIsInstance(vector.new_vector_search_scan(),
PrimaryKeyVectorScan)
- self.assertIsInstance(vector.new_vector_search_read(),
PrimaryKeyVectorRead)
- self.assertIsInstance(full_text.new_full_text_scan(),
PrimaryKeyFullTextScan)
- self.assertIsInstance(full_text.new_full_text_read(),
PrimaryKeyFullTextRead)
+ # A schema-only table selects the PK readers without a native table.
+ self.assertIsInstance(vector.new_vector_search_scan(),
PrimaryKeyVectorScan)
+ self.assertIsInstance(vector.new_vector_search_read(),
PrimaryKeyVectorRead)
+ self.assertIsInstance(full_text.new_full_text_scan(),
PrimaryKeyFullTextScan)
+ self.assertIsInstance(full_text.new_full_text_read(),
PrimaryKeyFullTextRead)
def test_primary_key_full_text_only_supports_fast_mode(self):
fields = [DataField(0, "id", AtomicType("INT")),
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index 9fa88a19e4..e915fd8a9d 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -1789,17 +1789,24 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(out['name'], ['old'])
self.assertEqual(out['age'], [99])
- def test_callable_value_rejected(self):
+ def test_callable_values_read_qualified_target_and_source(self):
target = self._create_table()
- with self.assertRaises(TypeError):
- merge_into(
- target=target,
- source=self._source(),
- catalog_options=self.catalog_options,
- on=['id'],
- when_matched=[WhenMatched.update({'name': lambda r: r})],
- num_partitions=_TEST_NUM_PARTITIONS,
- )
+ self._write(target, self._source())
+ merge_into(
+ target=target,
+ source=self._source(),
+ catalog_options=self.catalog_options,
+ on=['id'],
+ when_matched=[WhenMatched.update({
+ 'age': lambda rows: pc.add(rows['t.age'], rows['s.age']),
+ 'name': lambda rows: pc.utf8_upper(rows['t.name']),
+ })],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+ expected = self._source().sort_by('id').to_pydict()
+ actual = self._read_sorted(target)
+ self.assertEqual(actual['age'], [age * 2 for age in expected['age']])
+ self.assertEqual(actual['name'], [name.upper() for name in
expected['name']])
def test_source_missing_referenced_col(self):
target = self._create_table()
@@ -2530,6 +2537,7 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(result['num_matched'], 4)
self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99, 99])
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
def test_self_merge_rebases_staged_updates_after_compaction(self):
from pypaimon.data.generic_variant import GenericVariant
from pypaimon.data.variant_path import variant_get, variant_replace
@@ -3507,17 +3515,13 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
def test_self_merge_callable_validation(self):
target = self._create_table()
- with self.assertRaisesRegex(
- ValueError, 'Callable SET values require read_columns'):
- merge_into(
- target=target,
- source=target,
- catalog_options=self.catalog_options,
- on=['_ROW_ID'],
- when_matched=[WhenMatched.update({
- 'age': lambda rows: rows['age'],
- })],
- )
+ from pypaimon.ray.data_evolution_merge_into import _prepare
+ _, _, _, _, ctx = _prepare(
+ target, target, self.catalog_options,
+ [WhenMatched.update({'age': lambda rows: rows['t.age']})],
+ [], ['_ROW_ID'],
+ )
+ self.assertEqual(ctx.read_columns, ())
with self.assertRaisesRegex(
ValueError, 'read_columns requires a callable SET value'):
@@ -3543,20 +3547,6 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
})],
)
- source = self._create_table()
- with self.assertRaisesRegex(
- TypeError, 'callables are only supported for self-merge'):
- merge_into(
- target=target,
- source=source,
- catalog_options=self.catalog_options,
- on=['id'],
- read_columns=['age'],
- when_matched=[WhenMatched.update({
- 'age': lambda rows: rows['age'],
- })],
- )
-
def test_self_merge_callable_rejects_invalid_result(self):
from pypaimon.ray.data_evolution_merge_transform import (
_resolve_spec_array,
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 576947c526..13a94e9c68 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -935,7 +935,7 @@ class TableWriteTest(unittest.TestCase):
}, schema=pa_schema)
planner = PostponeBucketPlanner(
- table, known_num_buckets={}, postpone_row_counts={})
+ table, known_num_buckets={})
stats = planner.input_partition_stats(data)
self.assertEqual((1000, 32000), stats[()])
self.assertEqual(2, planner.plan(stats).num_buckets(()))
@@ -956,8 +956,7 @@ class TableWriteTest(unittest.TestCase):
}, schema=self.postpone_pa_schema)
planner = PostponeBucketPlanner(
table,
- known_num_buckets={('known',): 2},
- postpone_row_counts={},
+ known_num_buckets={('known',): 2}
)
self.assertEqual(
@@ -997,7 +996,7 @@ class TableWriteTest(unittest.TestCase):
}, schema=pa_schema)
planner = PostponeBucketPlanner(
- table, known_num_buckets={}, postpone_row_counts={})
+ table, known_num_buckets={})
self.assertEqual((1, 176), planner.input_partition_stats(data)[()])
def test_postpone_size_inference_supports_nested_vector(self):
@@ -1018,7 +1017,7 @@ class TableWriteTest(unittest.TestCase):
}, schema=pa_schema)
planner = PostponeBucketPlanner(
- table, known_num_buckets={}, postpone_row_counts={})
+ table, known_num_buckets={})
self.assertEqual((1, 80), planner.input_partition_stats(data)[()])
def test_postpone_default_bucket_function_matches_java(self):
@@ -1243,7 +1242,7 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(16, self._read_sorted(table, 'id').num_rows)
- def test_postpone_target_row_num_counts_existing_postpone_rows(self):
+ def test_postpone_target_row_num_uses_only_new_input(self):
from pypaimon.write.postpone_bucket import (
PostponeBucketPlanner,
)
@@ -1257,21 +1256,22 @@ class TableWriteTest(unittest.TestCase):
'postpone.batch-write-fixed-bucket.max-parallelism': 8,
},
)
- append_planner = PostponeBucketPlanner(
- table,
- known_num_buckets={},
- postpone_row_counts={('p',): 3},
- )
+ # Old staged rows must not influence the current batch's bucket count.
+ staged = self.pk_expected.set_column(
+ self.pk_expected.schema.get_field_index('dt'),
self.pk_expected.schema.field('dt'),
+ pa.array(['p'] * self.pk_expected.num_rows))
+ pending = self._commit_arrow(table, staged)
+ self.assertEqual({-2}, {message.bucket for message in pending})
+ append_planner = PostponeBucketPlanner(table)
append_plan = append_planner.plan({('p',): (1, 10)})
- self.assertEqual(2, append_plan.num_buckets(('p',)))
+ self.assertEqual(1, append_plan.num_buckets(('p',)))
overwrite_planner = PostponeBucketPlanner(
table,
- known_num_buckets={},
- postpone_row_counts={('p',): 3},
+ known_num_buckets={}
)
overwrite_plan = overwrite_planner.plan(
- {('p',): (1, 10)}, include_postpone_rows=False
+ {('p',): (1, 10)}
)
self.assertEqual(1, overwrite_plan.num_buckets(('p',)))
diff --git a/paimon-python/pypaimon/write/native_blob_rows.py
b/paimon-python/pypaimon/write/native_blob_rows.py
index 7505e82272..7feb12c08e 100644
--- a/paimon-python/pypaimon/write/native_blob_rows.py
+++ b/paimon-python/pypaimon/write/native_blob_rows.py
@@ -47,6 +47,7 @@ class NativeBlobRows:
def __init__(self, fallback=None):
self.readers = {}
+ self._consumed = set()
self.fallback = fallback
def _supports_uri(self, uri):
@@ -62,9 +63,20 @@ class NativeBlobRows:
def create(self, uri):
if uri in self.readers:
+ self._consumed.add(uri)
return self.readers[uri]
return self.fallback.create(uri)
+ def release_consumed(self):
+ """Release inputs consumed by core, retaining unplanned partition
rows."""
+ for uri in self._consumed:
+ self.readers.pop(uri, None)
+ self._consumed.clear()
+
+ def clear(self):
+ self.readers.clear()
+ self._consumed.clear()
+
def to_batch(self, table, rows, names):
"""Transport existing row values; core owns physical input
validation."""
descriptor_fields = table.options.blob_descriptor_fields()
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 623d067daf..56d6d3db64 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -17,6 +17,8 @@
"""Optional native commits using the Java CommitMessage v14 bridge."""
+import json
+
from pypaimon.common.json_util import JSON
from pypaimon.read.native_plan import (
native_runtime_available, _catalog_context_options, _catalog_metastore,
_option_value_to_string,
@@ -45,7 +47,7 @@ def native_messages_supported(table, messages) -> bool:
return True
-def create_native_commit(table, commit_user, overwrite_partition=None):
+def create_native_commit(table, commit_user, overwrite_partition=None, *,
batch=False):
"""Return a native committer only when its publication protocol matches
Python."""
if not _rest_catalog_supported(table) or not native_runtime_available():
return None
@@ -59,10 +61,11 @@ def create_native_commit(table, commit_user,
overwrite_partition=None):
native_table = create_native_write_table(table)
if native_table is None:
return None
- if overwrite_partition is not None:
- return (native_table.new_batch_write_builder()
- ._with_commit_user(commit_user)
- .with_overwrite(overwrite_partition).new_commit())
+ if batch or overwrite_partition is not None:
+ builder =
native_table.new_batch_write_builder()._with_commit_user(commit_user)
+ if overwrite_partition is not None:
+ builder = builder.with_overwrite(overwrite_partition)
+ return builder.new_commit()
# Append commits use the stream committer with the Python writer's
identity;
# Python enforces each mode's lifecycle and empty-commit rules.
return
native_table.new_stream_write_builder().with_commit_user(commit_user).new_commit()
@@ -72,7 +75,7 @@ def _rest_catalog_supported(table):
from pypaimon.catalog.catalog_environment import CatalogEnvironment
from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
from pypaimon.filesystem.caching_file_io import CachingFileIO
- from pypaimon.filesystem.local_file_io import LocalFileIO
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO, LocalFileIO
from pypaimon.filesystem.oss_file_io import OssFileIO
from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
@@ -86,26 +89,47 @@ def _rest_catalog_supported(table):
file_io = file_io._delegate
return (type(table) is FileStoreTable
and type(environment) is CatalogEnvironment
- and type(file_io) in (LocalFileIO, PyArrowFileIO, OssFileIO,
+ and type(file_io) in (LocalFileIO, FuseLocalFileIO, PyArrowFileIO,
OssFileIO,
ResolvingFileIO, RESTTokenFileIO)
and environment.supports_version_management
- and environment.uuid is not None
+ and _rest_source_uuid(table) is not None
and context is not None
and context.options is not None
and all(getattr(context, attr, None) is None for attr in (
'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')))
+def _rest_source_uuid(table):
+ """Identity of the catalog table used to reconstruct a copied FUSE view."""
+ if table.catalog_environment.uuid is not None:
+ return table.catalog_environment.uuid
+ from pypaimon.read.native_plan import _resolved_rest_table_response,
_validate_native_fuse_binding
+ try:
+ original = _validate_native_fuse_binding(table)
+ if original is None or original == table.identifier:
+ return None
+ response = _resolved_rest_table_response(table, original)
+ uuid = json.loads(response).get('id') if response is not None else None
+ return uuid if isinstance(uuid, str) and uuid else None
+ except (ValueError, RuntimeError):
+ return None
+
+
def _native_rest_table(table, schema_json):
from pypaimon_rust.datafusion import PaimonCatalog as NativeCatalog
+ from pypaimon.read.native_plan import _validate_native_fuse_binding
+ identifier = _validate_native_fuse_binding(table) or table.identifier
catalog_options = _catalog_context_options(table)
catalog_options['metastore'] = 'rest'
native_table = NativeCatalog(catalog_options,
config_required=False).get_table((
- table.identifier.get_database_name(),
table.identifier.get_object_name()))
+ identifier.get_database_name(), identifier.get_object_name()))
if (native_table.location() != table.table_path
- or native_table.rest_table_uuid() != table.catalog_environment.uuid
- or native_table.branch() != table.current_branch()):
+ or native_table.rest_table_uuid() != _rest_source_uuid(table)):
+ return None
+ if identifier != table.identifier:
+ native_table = native_table.copy({'branch': table.current_branch()})
+ if native_table.location() != table.table_path or native_table.branch() !=
table.current_branch():
return None
return native_table.copy_with_resolved_schema(schema_json)
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index b5f7c5236d..3338ccefd3 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -30,6 +30,15 @@ from pypaimon.write.native_update import (
from pypaimon.write.native_write import _native_partition_types_supported
+def _callable_assignment(function, on_pairs):
+ from pypaimon.ray.data_evolution_merge_transform import
_merge_callable_input
+
+ def evaluate(batch):
+ return function(_merge_callable_input(batch, on_pairs))
+
+ return evaluate
+
+
def create_native_merge_into(table, source, on, matched, not_matched,
commit_user, commit_identifier):
from pypaimon.table.data_evolution_merge_into import _is_self_merge,
_is_table_like, _prepare
from pypaimon.ray.data_evolution_merge_into import _normalize_on,
_union_update_cols
@@ -90,8 +99,6 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
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()):
- return None
if not _native_update_columns_supported(table,
_union_update_cols(matched)):
return None
for target, origin in zip(target_keys, source_keys):
@@ -109,6 +116,9 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
values.append((name, 'target', value.column))
elif isinstance(value, LiteralValue):
values.append((name, 'literal', value.value))
+ elif callable(value) and not isinstance(value, type):
+ values.append((name, 'callable', _callable_assignment(
+ value, list(zip(source_keys, target_keys)))))
else:
return None
condition = None if clause.condition is None else
dict(sql=rewrite_condition(clause.condition))
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 99c3aa49c8..da7fb72c90 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -437,7 +437,7 @@ class NativeTableUpsert:
self.writer.upsert_by_arrow_with_key(self.data, self.keys))
finally:
if self.blob_rows is not None:
- self.blob_rows.readers.clear()
+ self.blob_rows.clear()
class NativePredicateTableUpdate:
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 57dce53372..e26a4d87c3 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -104,6 +104,7 @@ class NativeTableWrite:
self._blob_uri_reader_factory = None
self._restore_snapshot_id = restore_snapshot_id
self._blob_rows = None
+ self._defer_blob_input = False
if any(is_blob_file_field(field) for field in
table.table_schema.fields):
from pypaimon.write.native_blob_rows import NativeBlobRows
self._blob_rows = NativeBlobRows()
@@ -260,7 +261,9 @@ class NativeTableWrite:
# Core externalizes Blob values during write, before buffering PK
# records. Their live Python objects need not survive the call.
if self._blob_rows is not None:
- self._blob_rows.readers.clear()
+ self._blob_rows.release_consumed()
+ if not self._defer_blob_input:
+ self._blob_rows.clear()
def prepare_commit(self, commit_identifier=None):
if self._python_writer is not None:
@@ -274,7 +277,11 @@ class NativeTableWrite:
else:
if commit_identifier is not None:
raise TypeError('BatchTableWrite.prepare_commit accepts no
identifier')
- messages = self._native_writer.prepare_commit()
+ try:
+ messages = self._native_writer.prepare_commit()
+ finally:
+ if self._blob_rows is not None and self._defer_blob_input:
+ self._blob_rows.clear()
return from_native_commit_messages(self.table, messages)
def close(self):
@@ -284,7 +291,7 @@ class NativeTableWrite:
self._native_writer.close()
self._native_writer = None
if self._blob_rows is not None:
- self._blob_rows.readers.clear()
+ self._blob_rows.clear()
def abort(self):
if self._python_writer is not None:
@@ -306,6 +313,9 @@ class NativePostponeFixedBucketTableWrite(NativeTableWrite):
def __init__(self, table, commit_user, static_partition, bucket_plan,
native_writer):
super().__init__(table, commit_user, static_partition, False,
native_writer)
self._bucket_plan = bucket_plan
+ # Automatic planning stages input until all partition sizes are known.
+ self._defer_blob_input = (bucket_plan is None
+ and
table.options.postpone_default_bucket_num() is None)
def _new_python_writer(self):
from pypaimon.write.postpone_batch_table_write import
PostponeFixedBucketBatchTableWrite
diff --git a/paimon-python/pypaimon/write/postpone_batch_table_write.py
b/paimon-python/pypaimon/write/postpone_batch_table_write.py
index f15f0bafb5..e1171f76ea 100644
--- a/paimon-python/pypaimon/write/postpone_batch_table_write.py
+++ b/paimon-python/pypaimon/write/postpone_batch_table_write.py
@@ -50,14 +50,10 @@ class PostponeFixedBucketWriteBuilder(BatchWriteBuilder):
return self
def new_write(self):
- # Automatic size estimation still belongs to the Python coordinator.
- # Native direct writes accept a resolved plan or Java's explicit
default.
- if (self._bucket_plan is not None
- or self.table.options.postpone_default_bucket_num() is not
None):
- native = self._native_write(
- self.static_partition, fixed_bucket=True,
bucket_plan=self._bucket_plan)
- if native is not None:
- return native
+ native = self._native_write(
+ self.static_partition, fixed_bucket=True,
bucket_plan=self._bucket_plan)
+ if native is not None:
+ return native
return PostponeFixedBucketBatchTableWrite(
self.table,
self.commit_user,
@@ -190,7 +186,6 @@ class PostponeFixedBucketBatchTableWrite(BatchTableWrite):
self._bucket_plan = self._planner.plan(
partition_stats,
- include_postpone_rows=self.static_partition is None,
)
self.row_key_extractor.with_bucket_plan(self._bucket_plan)
inputs = self._pending_inputs
diff --git a/paimon-python/pypaimon/write/postpone_bucket.py
b/paimon-python/pypaimon/write/postpone_bucket.py
index ca6c9a8100..d2dac9a6e9 100644
--- a/paimon-python/pypaimon/write/postpone_bucket.py
+++ b/paimon-python/pypaimon/write/postpone_bucket.py
@@ -220,7 +220,6 @@ class PostponeBucketPlanner:
self,
table,
known_num_buckets=None,
- postpone_row_counts=None,
):
options = table.options
if options.bucket() != BucketMode.POSTPONE_BUCKET.value:
@@ -273,13 +272,8 @@ class PostponeBucketPlanner:
self._partition_keys = list(table.partition_keys)
self._field_dict = dict(table.field_dict)
if known_num_buckets is None:
- known_num_buckets, loaded_postpone_counts = (
- self._load_bucket_metadata(table)
- )
- if postpone_row_counts is None:
- postpone_row_counts = loaded_postpone_counts
+ known_num_buckets = self._load_bucket_metadata(table)
self._known_num_buckets = dict(known_num_buckets)
- self._postpone_row_counts = dict(postpone_row_counts or {})
@staticmethod
def _load_bucket_metadata(table):
@@ -291,14 +285,9 @@ class PostponeBucketPlanner:
)
known = {}
- postpone_counts = {}
for entry in entries:
partition = tuple(entry.partition.values)
- if entry.bucket == BucketMode.POSTPONE_BUCKET.value:
- postpone_counts[partition] = (
- postpone_counts.get(partition, 0) + entry.file.row_count
- )
- elif entry.bucket >= 0 and entry.total_buckets > 0:
+ if entry.bucket >= 0 and entry.total_buckets > 0:
previous = known.get(partition)
if previous is not None and previous != entry.total_buckets:
raise RuntimeError(
@@ -306,7 +295,7 @@ class PostponeBucketPlanner:
.format(partition, previous, entry.total_buckets)
)
known[partition] = entry.total_buckets
- return known, postpone_counts
+ return known
def current_plan(self) -> PostponeBucketPlan:
return PostponeBucketPlan(self._known_num_buckets)
@@ -340,7 +329,6 @@ class PostponeBucketPlanner:
def plan(
self,
partition_stats: Dict[Tuple, Tuple[int, int]],
- include_postpone_rows: bool = True,
) -> PostponeBucketPlan:
for partition, (row_count, data_size) in partition_stats.items():
partition = tuple(partition)
@@ -349,13 +337,8 @@ class PostponeBucketPlanner:
if self.default_bucket_num is not None:
self._known_num_buckets[partition] = self.default_bucket_num
continue
- postpone_rows = (
- self._postpone_row_counts.get(partition, 0)
- if include_postpone_rows
- else 0
- )
if self.target_row_num_per_bucket is not None:
- total_rows = row_count + postpone_rows
+ total_rows = row_count
num_buckets = max(
1,
_ceil_div(
@@ -363,15 +346,12 @@ class PostponeBucketPlanner:
)
else:
estimated_size = data_size
- if postpone_rows and row_count:
- estimated_size = _ceil_div(
- data_size * (row_count + postpone_rows), row_count)
num_buckets = max(
1,
_ceil_div(
estimated_size, self.target_size_per_bucket),
)
self._known_num_buckets[partition] = min(
- num_buckets, self.max_num_buckets
+ 1 << (num_buckets - 1).bit_length(), self.max_num_buckets
)
return self.current_plan()
diff --git a/paimon-python/pypaimon/write/ray_datasink.py
b/paimon-python/pypaimon/write/ray_datasink.py
index edfdcb90ba..af8322cb40 100644
--- a/paimon-python/pypaimon/write/ray_datasink.py
+++ b/paimon-python/pypaimon/write/ray_datasink.py
@@ -346,9 +346,6 @@ def write_paimon_dataset(
)
plan = planner.plan(
partition_stats,
- include_postpone_rows=not (
- overwrite or static_partition is not None
- ),
)
return _write_primary_key_groups(
dataset,
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index 9819141476..8ae28d33a9 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -53,6 +53,7 @@ class TableCommit:
self._commit_callbacks: List[CommitCallback] = []
self._native_commit = None
+ self._native_commit_is_batch = False
self._native_callbacks_registered = False
self.file_store_commit = FileStoreCommit(
snapshot_commit, table, commit_user,
@@ -111,11 +112,14 @@ class TableCommit:
native, messages = prepared
# Mutation is deliberately outside the fallback boundary:
# an exception can mean the snapshot was already published.
- native.commit(commit_identifier, messages, **native_kwargs)
+ if self._native_commit_is_batch:
+ native.commit(messages, **native_kwargs)
+ else:
+ native.commit(commit_identifier, messages, **native_kwargs)
return
self.file_store_commit.commit(**commit_kwargs)
- def _prepare_native_commit(self, messages):
+ def _prepare_native_commit(self, messages, *, batch=False):
if not self.table.options.native_commit_enabled():
return None
try:
@@ -124,9 +128,17 @@ class TableCommit:
to_native_commit_messages)
if not native_messages_supported(self.table, messages):
return None
- if self._native_commit is None:
- self._native_commit = create_native_commit(
- self.table, self.commit_user, self.overwrite_partition)
+ if self._native_commit is None or (batch and not
self._native_commit_is_batch):
+ native = create_native_commit(
+ self.table, self.commit_user, self.overwrite_partition,
+ **({'batch': True} if batch else {}))
+ if native is None:
+ return None
+ if self._native_commit is not None:
+ self._native_commit.close()
+ self._native_commit = native
+ self._native_commit_is_batch = batch or
self.overwrite_partition is not None
+ self._native_callbacks_registered = False
if self._native_commit is None:
return None
native_messages = to_native_commit_messages(self.table, messages)
@@ -185,10 +197,17 @@ class BatchTableCommit(TableCommit):
def truncate_table(self) -> None:
"""Truncate the entire table, deleting all data."""
self._check_committed()
+ prepared = self._prepare_native_commit([], batch=True)
+ if prepared is not None:
+ prepared[0].truncate_table()
+ return
self.file_store_commit.truncate_table(BATCH_COMMIT_IDENTIFIER)
def truncate_partitions(self, partitions: List[Dict[str, str]]) -> None:
- self._check_committed()
+ prepared = self._prepare_native_commit([], batch=True)
+ if prepared is not None:
+ prepared[0].truncate_partitions(partitions)
+ return
self.file_store_commit.drop_partitions(partitions,
BATCH_COMMIT_IDENTIFIER)
def _check_committed(self):
