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 bab76b6198 [python] Enable native Paimon branch writes (#10385)
bab76b6198 is described below
commit bab76b6198509c2f6d5413f3320d967b1fa7620a
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 5 09:56:45 2026 +0800
[python] Enable native Paimon branch writes (#10385)
---
paimon-python/pypaimon/api/rest_api.py | 3 +-
paimon-python/pypaimon/api/rest_util.py | 10 +-
paimon-python/pypaimon/branch/branch_manager.py | 5 +-
.../pypaimon/branch/filesystem_branch_manager.py | 4 +-
.../pypaimon/tests/api/test_branch_dto_serde.py | 4 +-
.../pypaimon/tests/api/test_partition_dto_serde.py | 2 +-
.../pypaimon/tests/api/test_tag_dto_serde.py | 4 +-
.../pypaimon/tests/branch_manager_test.py | 4 +-
.../pypaimon/tests/native_branch_write_test.py | 481 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_commit_test.py | 5 +-
paimon-python/pypaimon/tests/rest/api_test.py | 20 +
paimon-python/pypaimon/tests/rest/rest_server.py | 82 +++-
paimon-python/pypaimon/write/native_commit.py | 9 +-
13 files changed, 589 insertions(+), 44 deletions(-)
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index 608b195974..b6ee8f4144 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -789,7 +789,8 @@ class RESTApi:
if not table_name or not table_name.strip():
raise ValueError("Table name cannot be None")
- return database_name.strip(), table_name.strip()
+ # Validation must not rename a resource, including a branch suffix.
+ return database_name, table_name
from pypaimon.catalog.catalog_exception import IllegalArgumentError
diff --git a/paimon-python/pypaimon/api/rest_util.py
b/paimon-python/pypaimon/api/rest_util.py
index 2ccb406406..04a02ba531 100644
--- a/paimon-python/pypaimon/api/rest_util.py
+++ b/paimon-python/pypaimon/api/rest_util.py
@@ -16,7 +16,7 @@
# under the License.
from typing import Dict, Optional
-from urllib.parse import unquote
+from urllib.parse import quote_plus, unquote_plus
from pypaimon.common.options import Options
@@ -24,14 +24,14 @@ from pypaimon.common.options import Options
class RESTUtil:
@staticmethod
def encode_string(value: str) -> str:
- import urllib.parse
-
- return urllib.parse.quote(value)
+ # Java URLEncoder uses form encoding, including '+' for spaces and
+ # escaping '/' so one resource name stays one path segment.
+ return quote_plus(value, safe="*").replace("~", "%7E")
@staticmethod
def decode_string(encoded: str) -> str:
"""Decode URL-encoded string"""
- return unquote(encoded)
+ return unquote_plus(encoded)
@staticmethod
def extract_prefix_map(
diff --git a/paimon-python/pypaimon/branch/branch_manager.py
b/paimon-python/pypaimon/branch/branch_manager.py
index b7e0b443d7..ae0d160f17 100644
--- a/paimon-python/pypaimon/branch/branch_manager.py
+++ b/paimon-python/pypaimon/branch/branch_manager.py
@@ -143,7 +143,8 @@ class BranchManager:
"""
if not branch or not branch.strip():
return DEFAULT_MAIN_BRANCH
- return branch.strip()
+ # Java preserves non-blank names; trimming would alias branch
directories.
+ return branch
@staticmethod
def is_main_branch(branch: str) -> bool:
@@ -175,7 +176,7 @@ class BranchManager:
)
if not branch_name or not branch_name.strip():
raise ValueError("Branch name is blank.")
- if branch_name.strip().isdigit():
+ if branch_name.isdigit():
raise ValueError(
f"Branch name cannot be pure numeric string but is
'{branch_name}'."
)
diff --git a/paimon-python/pypaimon/branch/filesystem_branch_manager.py
b/paimon-python/pypaimon/branch/filesystem_branch_manager.py
index f191d29a63..855e30603e 100644
--- a/paimon-python/pypaimon/branch/filesystem_branch_manager.py
+++ b/paimon-python/pypaimon/branch/filesystem_branch_manager.py
@@ -61,9 +61,7 @@ class FileSystemBranchManager(BranchManager):
@staticmethod
def _normalize_branch(branch: str) -> str:
"""Normalize branch name."""
- if not branch or branch == "main":
- return "main"
- return branch
+ return BranchManager.normalize_branch(branch)
def _branch_directory(self) -> str:
"""Return the root directory of branch."""
diff --git a/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
b/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
index 58d7139bcf..a211c7af69 100644
--- a/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
+++ b/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
@@ -124,10 +124,10 @@ class ResourcePathsBranchesTest(unittest.TestCase):
def test_branch_url_url_encodes_branch_name(self):
# RESTUtil.encode_string escapes characters that are not URL-safe;
- # a space round-trips to ``%20``. Mirrors the existing tag-name path.
+ # a space round-trips to ``+``. Mirrors the existing tag-name path.
self.assertEqual(
self.paths.branch("db", "tbl", "release 1.0"),
- "/v1/mock/databases/db/tables/tbl/branches/release%201.0",
+ "/v1/mock/databases/db/tables/tbl/branches/release+1.0",
)
diff --git a/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
b/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
index 7b5ec38f40..bbb91ab78d 100644
--- a/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
+++ b/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
@@ -96,7 +96,7 @@ class ResourcePathsPartitionsTest(unittest.TestCase):
paths = ResourcePaths(prefix="mock")
self.assertEqual(
paths.partitions("my db", "my tbl"),
- "/v1/mock/databases/my%20db/tables/my%20tbl/partitions",
+ "/v1/mock/databases/my+db/tables/my+tbl/partitions",
)
if __name__ == "__main__":
diff --git a/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
b/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
index 421dbf8d7c..ddcfbb1be2 100644
--- a/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
+++ b/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
@@ -132,10 +132,10 @@ class ResourcePathsTagsTest(unittest.TestCase):
def test_single_tag_url_url_encodes_tag_name(self):
# RESTUtil.encode_string escapes characters that are not URL-safe.
- # A space round-trips to ``%20``; this matches Java's ``RESTUtil``.
+ # A space round-trips to ``+``; this matches Java's ``RESTUtil``.
paths = ResourcePaths(prefix="mock")
url = paths.tag("db", "tbl", "release 1.0")
- self.assertEqual(url,
"/v1/mock/databases/db/tables/tbl/tags/release%201.0")
+ self.assertEqual(url,
"/v1/mock/databases/db/tables/tbl/tags/release+1.0")
if __name__ == "__main__":
diff --git a/paimon-python/pypaimon/tests/branch_manager_test.py
b/paimon-python/pypaimon/tests/branch_manager_test.py
index d51d69b188..dc70797b76 100644
--- a/paimon-python/pypaimon/tests/branch_manager_test.py
+++ b/paimon-python/pypaimon/tests/branch_manager_test.py
@@ -44,7 +44,7 @@ class BranchManagerTest(unittest.TestCase):
self.assertEqual(BranchManager.normalize_branch(" "),
DEFAULT_MAIN_BRANCH)
self.assertEqual(BranchManager.normalize_branch("main"), "main")
self.assertEqual(BranchManager.normalize_branch("feature"), "feature")
- self.assertEqual(BranchManager.normalize_branch(" feature "),
"feature")
+ self.assertEqual(BranchManager.normalize_branch(" feature "), "
feature ")
def test_branch_path(self):
"""Test branch_path method."""
@@ -87,6 +87,8 @@ class BranchManagerTest(unittest.TestCase):
BranchManager.validate_branch("feature")
BranchManager.validate_branch("develop")
BranchManager.validate_branch("feature-branch-123")
+ BranchManager.validate_branch(" main ")
+ BranchManager.validate_branch(" 123 ")
def test_fast_forward_validate_to_main(self):
"""Test fast_forward_validate rejects fast-forward to main."""
diff --git a/paimon-python/pypaimon/tests/native_branch_write_test.py
b/paimon-python/pypaimon/tests/native_branch_write_test.py
new file mode 100644
index 0000000000..38de93fd4e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_branch_write_test.py
@@ -0,0 +1,481 @@
+# 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.
+
+"""Branch writes follow Java's independent snapshot and shared-data layout."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.identifier import Identifier
+from pypaimon.common.options.core_options import MergeEngine
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = pytest.mark.native_plan
+_SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.int32()), ('p',
pa.string())])
+
+
+def _table(tmp_path, rest_catalog, rest, options=None, primary_keys=None):
+ catalog = rest_catalog if rest else CatalogFactory.create({'warehouse':
str(tmp_path)})
+ catalog.create_database('default', True)
+ settings = {'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'commit.native.enabled': 'true', 'manifest.sidecar.enabled':
'true',
+ 'bucket': '1' if primary_keys else '-1'}
+ settings.update(options or {})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, primary_keys=primary_keys or [], partition_keys=['p'],
options=settings), False)
+ return catalog, catalog.get_table('default.t')
+
+
+def _prepare(builder, rows, checkpoint=None):
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(rows if isinstance(rows, pa.Table)
+ else pa.Table.from_pylist(rows, schema=_SCHEMA))
+ assert writer._python_writer is None
+ return writer.prepare_commit() if checkpoint is None else
writer.prepare_commit(checkpoint)
+ finally:
+ writer.close()
+
+
+def _commit(builder, messages, checkpoint=None, overwrite=False):
+ commit = builder.new_commit()
+ try:
+ def publish():
+ if checkpoint is None:
+ commit.commit(messages)
+ else:
+ commit.commit(messages, checkpoint)
+ if builder.table.catalog_environment.supports_version_management:
+ with patch.object(commit.file_store_commit, 'overwrite' if
overwrite else 'commit',
+ side_effect=AssertionError('Python commit
fallback')):
+ publish()
+ else:
+ publish()
+ finally:
+ commit.close()
+
+
+def _write(table, rows):
+ builder = (table.new_postpone_fixed_bucket_write_builder()
+ if table.options.bucket() == -2 else
table.new_batch_write_builder())
+ _commit(builder, _prepare(builder, rows))
+
+
+def _branch(catalog, table, from_tag=True, name='dev'):
+ if from_tag:
+ catalog.create_tag(table.identifier, 'base')
+ catalog.create_branch(table.identifier, name, tag_name='base' if from_tag
else None)
+ return catalog.get_table(Identifier('default', 't', branch=name))
+
+
+def _rows(table, native):
+ builder = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled':
str(native).lower()}).new_read_builder()
+ scan = builder.new_scan()
+ splits = (scan.plan_for_write() if table.is_primary_key_table
+ and table.options.merge_engine() == MergeEngine.FIRST_ROW else
scan.plan()).splits()
+ reader = builder.new_read()
+ if native:
+ with patch.object(reader, '_create_split_read',
side_effect=AssertionError('Python read fallback')):
+ rows = reader.to_arrow(splits).to_pylist()
+ else:
+ rows = reader.to_arrow(splits).to_pylist()
+ return sorted(rows, key=lambda row: (row['p'], row['id'], row['value']))
+
+
+def _assert_rows(table, expected):
+ for native in (False, True):
+ assert _rows(table, native) == expected
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('from_tag', [False, True], ids=['empty', 'tagged'])
+def test_native_branch_append_isolates_main(tmp_path, native_rest_catalog,
rest, from_tag):
+ catalog, main = _table(tmp_path, native_rest_catalog, rest)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [seed])
+ branch = _branch(catalog, main, from_tag)
+ _write(main, [{'id': 2, 'value': 200, 'p': 'a'}])
+ _write(branch, [{'id': 3, 'value': 30, 'p': 'b'}])
+ _assert_rows(main, [seed, {'id': 2, 'value': 200, 'p': 'a'}])
+ _assert_rows(branch, ([seed] if from_tag else []) + [{'id': 3, 'value':
30, 'p': 'b'}])
+ assert main.snapshot_manager().get_latest_snapshot().id == 2
+ assert branch.snapshot_manager().get_latest_snapshot().id == (2 if
from_tag else 1)
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('engine,branch_value,main_value', [
+ ('deduplicate', 20, 100), ('first-row', 10, 10),
+ ('partial-update', 20, 100), ('aggregation', 30, 110),
+])
+def test_native_branch_primary_key_merge_engines(
+ tmp_path, native_rest_catalog, rest, engine, branch_value, main_value):
+ options = {'merge-engine': engine}
+ if engine == 'aggregation':
+ options['fields.value.aggregate-function'] = 'sum'
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, options, ['p',
'id'])
+ _write(main, [{'id': 1, 'value': 10, 'p': 'a'}])
+ branch = _branch(catalog, main)
+ _write(main, [{'id': 1, 'value': 100, 'p': 'a'}])
+ _write(branch, [{'id': 1, 'value': 20, 'p': 'a'}])
+ _assert_rows(main, [{'id': 1, 'value': main_value, 'p': 'a'}])
+ _assert_rows(branch, [{'id': 1, 'value': branch_value, 'p': 'a'}])
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('routing', ['dynamic', 'cross-partition', 'postpone'])
+def test_native_branch_bucket_routing(tmp_path, native_rest_catalog, rest,
routing):
+ options = {'bucket': '-2' if routing == 'postpone' else '-1',
+ 'dynamic-bucket.target-row-num': '1',
'postpone.default-bucket-num': '2'}
+ keys = ['id'] if routing == 'cross-partition' else ['p', 'id']
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, options, keys)
+ original = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [original])
+ branch = _branch(catalog, main)
+ _write(main, [{'id': 2, 'value': 200, 'p': 'b'}])
+ _write(branch, [{'id': 1, 'value': 20, 'p': 'b'}, {'id': 3, 'value': 30,
'p': 'b'}])
+ _assert_rows(main, [original, {'id': 2, 'value': 200, 'p': 'b'}])
+ expected = [] if routing == 'cross-partition' else [original]
+ _assert_rows(branch, expected + [
+ {'id': 1, 'value': 20, 'p': 'b'}, {'id': 3, 'value': 30, 'p': 'b'}])
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('data_evolution', [False, True], ids=['append',
'data-evolution'])
+def test_native_branch_overwrite(tmp_path, native_rest_catalog, rest,
data_evolution):
+ options = ({'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true'}
+ if data_evolution else {})
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, options)
+ original = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [original])
+ branch = _branch(catalog, main)
+ builder = branch.new_batch_write_builder().overwrite({'p': 'a'})
+ changed = {'id': 2, 'value': 20, 'p': 'a'}
+ _commit(builder, _prepare(builder, [changed]), overwrite=True)
+ _assert_rows(main, [original])
+ _assert_rows(branch, [changed])
+ assert main.snapshot_manager().get_latest_snapshot().id == 1
+ assert branch.snapshot_manager().get_latest_snapshot().id == 2
+ if data_evolution:
+ assert main.snapshot_manager().get_latest_snapshot().next_row_id == 1
+ assert branch.snapshot_manager().get_latest_snapshot().next_row_id == 2
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_stream_checkpoint_histories(tmp_path,
native_rest_catalog, rest):
+ catalog, main = _table(tmp_path, native_rest_catalog, rest,
primary_keys=['p', 'id'])
+ _write(main, [{'id': 1, 'value': 10, 'p': 'a'}])
+ branch = _branch(catalog, main)
+ for target, value in ((main, 100), (branch, 20)):
+ builder =
target.new_stream_write_builder()._with_commit_user('same-writer')
+ messages = _prepare(builder, [{'id': 1, 'value': value, 'p': 'a'}],
checkpoint=7)
+ _commit(builder, messages, checkpoint=7)
+ snapshot = target.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == 2
+ assert snapshot.commit_identifier == 7
+ _assert_rows(main, [{'id': 1, 'value': 100, 'p': 'a'}])
+ _assert_rows(branch, [{'id': 1, 'value': 20, 'p': 'a'}])
+
+
+def _update_value(table, value):
+ from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+ builder = table.new_batch_write_builder()
+ data = pa.table({'_ROW_ID': pa.array([0], type=pa.int64()),
+ 'value': pa.array([value], type=pa.int32())})
+ with patch.object(TableUpdateByRowId, 'update_columns',
+ side_effect=AssertionError('Python update fallback')):
+ messages =
builder.new_update().with_update_type(['value']).update_by_arrow_with_row_id(data)
+ assert messages
+ return builder, messages
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_row_id_conflicts_are_scoped(tmp_path,
native_rest_catalog, rest):
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'})
+ _write(main, [{'id': 1, 'value': 10, 'p': 'a'}])
+ branch = _branch(catalog, main)
+ main_builder, main_messages = _update_value(main, 100)
+ branch_builder, branch_messages = _update_value(branch, 20)
+ _commit(branch_builder, branch_messages)
+ _commit(main_builder, main_messages)
+ _assert_rows(main, [{'id': 1, 'value': 100, 'p': 'a'}])
+ _assert_rows(branch, [{'id': 1, 'value': 20, 'p': 'a'}])
+ stale_builder, stale_messages = _update_value(branch, 21)
+ fresh_builder, fresh_messages = _update_value(branch, 22)
+ _commit(fresh_builder, fresh_messages)
+ with pytest.raises(Exception, match='conflict|Conflict|overlap'):
+ _commit(stale_builder, stale_messages)
+ assert branch.snapshot_manager().get_latest_snapshot().id == 3
+ _assert_rows(branch, [{'id': 1, 'value': 22, 'p': 'a'}])
+ _assert_rows(main, [{'id': 1, 'value': 100, 'p': 'a'}])
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('from_tag', [False, True], ids=['empty', 'tagged'])
+def test_native_branch_deletion_vectors_and_row_id_allocation(
+ tmp_path, native_rest_catalog, rest, from_tag):
+ from pypaimon.write.table_delete import TableDeleteByRowId
+
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true'})
+ original = [{'id': 1, 'value': 10, 'p': 'a'}, {'id': 2, 'value': 20, 'p':
'a'}]
+ _write(main, original)
+ branch = _branch(catalog, main, from_tag)
+ if not from_tag:
+ _write(branch, original)
+ builder = branch.new_batch_write_builder()
+ with patch.object(TableDeleteByRowId, 'delete',
+ side_effect=AssertionError('Python delete fallback')):
+ messages = builder.new_update().delete_by_row_id([0])
+ _commit(builder, messages)
+ _assert_rows(main, original)
+ _assert_rows(branch, [original[1]])
+ assert branch.snapshot_manager().get_latest_snapshot().next_row_id == 2
+ third = {'id': 3, 'value': 30, 'p': 'a'}
+ _write(branch, [third])
+ _assert_rows(branch, [original[1], third])
+ assert branch.snapshot_manager().get_latest_snapshot().next_row_id == 3
+ assert main.snapshot_manager().get_latest_snapshot().next_row_id == 2
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_schema_evolution_is_independent(tmp_path,
native_rest_catalog, rest):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+
+ catalog, main = _table(tmp_path, native_rest_catalog, rest)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [seed])
+ branch = _branch(catalog, main)
+ catalog.alter_table(main.identifier, [SchemaChange.add_column('main_only',
AtomicType('INT'))], False)
+ main = catalog.get_table(main.identifier)
+ main_schema = _SCHEMA.append(pa.field('main_only', pa.int32()))
+ _write(main, pa.Table.from_pylist(
+ [{'id': 2, 'value': 200, 'p': 'a', 'main_only': 7}],
schema=main_schema))
+ _write(branch, [{'id': 3, 'value': 30, 'p': 'b'}])
+ assert main.snapshot_manager().get_latest_snapshot().schema_id == 1
+ assert branch.snapshot_manager().get_latest_snapshot().schema_id == 0
+ assert 'main_only' not in branch.field_names
+ catalog.alter_table(branch.identifier, [
+ SchemaChange.add_column('dev_only', AtomicType('STRING')),
+ SchemaChange.set_option('target-file-row-num', '1')], False)
+ branch = catalog.get_table(branch.identifier)
+ branch_schema = _SCHEMA.append(pa.field('dev_only', pa.string()))
+ _write(branch, pa.Table.from_pylist(
+ [{'id': 4, 'value': 40, 'p': 'b', 'dev_only': 'dev'}],
schema=branch_schema))
+ _assert_rows(branch, [dict(seed, dev_only=None),
+ {'id': 3, 'value': 30, 'p': 'b', 'dev_only': None},
+ {'id': 4, 'value': 40, 'p': 'b', 'dev_only': 'dev'}])
+ _assert_rows(main, [dict(seed, main_only=None),
+ {'id': 2, 'value': 200, 'p': 'a', 'main_only': 7}])
+ assert 'dev_only' not in main.field_names
+ assert branch.snapshot_manager().get_latest_snapshot().schema_id ==
branch.table_schema.id
+ assert main.snapshot_manager().get_latest_snapshot().schema_id == 1
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_predicate_upsert_and_batch_updates(tmp_path,
native_rest_catalog, rest):
+ from pypaimon.read.table_read import TableRead
+ from pypaimon.write.table_update import BatchTableUpdate
+ from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+ from pypaimon.write.table_delete import TableDeleteByRowId
+
+ catalog, main = _table(tmp_path, native_rest_catalog, rest, {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true', 'target-file-row-num': '1'})
+ original = [{'id': 1, 'value': 10, 'p': 'a'}, {'id': 2, 'value': 20, 'p':
'b'}]
+ _write(main, original)
+ branch = _branch(catalog, main)
+ predicate = branch.new_read_builder().new_predicate_builder().equal('id',
1)
+ builder = branch.new_batch_write_builder()
+ with patch.object(TableRead, 'to_arrow',
side_effect=AssertionError('Python predicate scan fallback')):
+ messages = builder.new_update().update_by_predicate(
+ predicate, {'value': lambda batch: pa.compute.add(batch['value'],
1)}, ['value'])
+ _commit(builder, messages)
+ builder = branch.new_batch_write_builder()
+ with patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('Python upsert fallback')):
+ messages =
builder.new_update().upsert_by_arrow_with_key(pa.Table.from_pylist([
+ {'id': 1, 'value': 12, 'p': 'a'}, {'id': 3, 'value': 30, 'p':
'c'}], schema=_SCHEMA), ['id'])
+ _commit(builder, messages)
+ # Partition groups may be published in either order. Match logical rows
+ # to their actual IDs rather than assuming a HashMap iteration order.
+ reader = branch.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ read = reader.new_read_builder().with_projection(['id', '_ROW_ID'])
+ ids = {row['id']: row['_ROW_ID'] for row in read.new_read().to_arrow(
+ read.new_scan().plan().splits()).to_pylist()}
+ builder = branch.new_batch_write_builder()
+ with patch.object(BatchTableUpdate, '_update_by_arrow_batches_with_row_id',
+ side_effect=AssertionError('Python grouped update
fallback')):
+ messages =
builder.new_update().with_update_type(['value']).update_by_arrow_batches_with_row_id([
+ pa.table({'_ROW_ID': [ids[1]], 'value': [13]}),
+ pa.table({'_ROW_ID': [ids[2]], 'value': [21]})])
+ _commit(builder, messages)
+ builder = branch.new_batch_write_builder()
+ with patch.object(TableDeleteByRowId, 'delete',
side_effect=AssertionError('Python delete fallback')):
+ messages = builder.new_update().delete_by_predicate(
+ branch.new_read_builder().new_predicate_builder().equal('id', 2))
+ _commit(builder, messages)
+ _assert_rows(main, original)
+ _assert_rows(branch, [{'id': 1, 'value': 13, 'p': 'a'}, {'id': 3, 'value':
30, 'p': 'c'}])
+ assert branch.snapshot_manager().get_latest_snapshot().next_row_id == 3
+ assert main.snapshot_manager().get_latest_snapshot().next_row_id == 2
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def
test_native_branch_blob_append_and_overwrite_leave_shared_payloads_readable(tmp_path,
native_rest_catalog, rest):
+ from pypaimon.schema.data_types import AtomicType, DataField
+
+ catalog = native_rest_catalog if rest else
CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ options = {'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'commit.native.enabled': 'true', 'manifest.sidecar.enabled':
'true',
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'blob.target-file-size': '1 B'}
+ fields = [DataField(0, 'id', AtomicType('INT')), DataField(1, 'value',
AtomicType('BLOB')),
+ DataField(2, 'p', AtomicType('STRING'))]
+ catalog.create_table('default.t', Schema(fields=fields,
partition_keys=['p'], options=options), False)
+ main = catalog.get_table('default.t')
+ schema = pa.schema([('id', pa.int32()), ('value', pa.large_binary()),
('p', pa.string())])
+ seed = {'id': 1, 'value': b'original', 'p': 'a'}
+ _write(main, pa.Table.from_pylist([seed], schema=schema))
+ branch = _branch(catalog, main)
+ builder = branch.new_batch_write_builder()
+ added = {'id': 2, 'value': b'branch', 'p': 'a'}
+ messages = _prepare(builder, pa.Table.from_pylist([added], schema=schema))
+ assert any(file.file_name.endswith('.blob') for message in messages for
file in message.new_files)
+ _commit(builder, messages)
+ assert _rows(main, True) == [seed]
+ assert _rows(branch, True) == [seed, added]
+ builder = branch.new_batch_write_builder().overwrite({'p': 'a'})
+ replacement = {'id': 2, 'value': b'replacement', 'p': 'a'}
+ _commit(builder, _prepare(builder, pa.Table.from_pylist([replacement],
schema=schema)), overwrite=True)
+ assert _rows(branch, True) == [replacement]
+ assert _rows(main, True) == [seed]
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_abort_removes_only_unpublished_output(tmp_path,
native_rest_catalog, rest):
+ catalog, main = _table(tmp_path, native_rest_catalog, rest)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [seed])
+ branch = _branch(catalog, main)
+ builder = branch.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 2, 'value': 20, 'p': 'a'}])
+ paths = [file.file_path for message in messages for file in
message.new_files]
+ assert paths and all(branch.file_io.exists(path) for path in paths)
+ commit = builder.new_commit()
+ try:
+ if rest:
+ with patch.object(commit.file_store_commit, 'abort',
+ side_effect=AssertionError('Python abort
fallback')):
+ commit.abort(messages)
+ else:
+ commit.abort(messages)
+ finally:
+ commit.close()
+ assert all(not branch.file_io.exists(path) for path in paths)
+ assert branch.snapshot_manager().get_latest_snapshot().id == 1
+ _assert_rows(main, [seed])
+ _assert_rows(branch, [seed])
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
+def test_native_branch_lifecycle_preserves_histories(tmp_path,
native_rest_catalog, rest):
+ from pypaimon.catalog.catalog_exception import TableNotExistException
+
+ catalog, main = _table(tmp_path, native_rest_catalog, rest)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ added = {'id': 2, 'value': 20, 'p': 'a'}
+ _write(main, [seed])
+ branch = _branch(catalog, main)
+ _write(branch, [added])
+ catalog.rename_branch(main.identifier, 'dev', 'renamed')
+ catalog.rename_table(main.identifier, Identifier('default', 'renamed'))
+ main = catalog.get_table('default.renamed')
+ branch = catalog.get_table(Identifier('default', 'renamed',
branch='renamed'))
+ _assert_rows(main, [seed])
+ _assert_rows(branch, [seed, added])
+ assert catalog.list_branches(main.identifier) == ['renamed']
+ assert catalog.get_tag(main.identifier, 'base').snapshot.id == 1
+ with pytest.raises(TableNotExistException):
+ catalog.get_table('default.t')
+ third = {'id': 3, 'value': 30, 'p': 'b'}
+ _write(branch, [third])
+ _assert_rows(branch, [seed, added, third])
+ catalog.drop_branch(main.identifier, 'renamed')
+ assert catalog.list_branches(main.identifier) == []
+ _assert_rows(main, [seed])
+ catalog.create_branch(main.identifier, 'renamed')
+ empty = catalog.get_table(Identifier('default', 'renamed',
branch='renamed'))
+ _assert_rows(empty, [])
+ _write(empty, [added])
+ _assert_rows(empty, [added])
+ assert empty.snapshot_manager().get_latest_snapshot().id == 1
+ _assert_rows(main, [seed])
+
+
[email protected]('name', ['main', '', ' ', '123'])
+def test_native_rest_branch_invalid_names_preserve_main(tmp_path,
native_rest_catalog, name):
+ from pypaimon.api.rest_exception import BadRequestException
+ from pypaimon.catalog.catalog_exception import IllegalArgumentError
+
+ catalog, main = _table(tmp_path, native_rest_catalog, True)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [seed])
+ for operation in (lambda: catalog.create_branch(main.identifier, name),
+ lambda: catalog.drop_branch(main.identifier, name)):
+ with pytest.raises((BadRequestException, IllegalArgumentError,
ValueError)):
+ operation()
+ _assert_rows(main, [seed])
+ branch = _branch(catalog, main)
+ with pytest.raises((BadRequestException, IllegalArgumentError,
ValueError)):
+ catalog.rename_branch(main.identifier, 'dev', name)
+ _assert_rows(branch, [seed])
+ _assert_rows(main, [seed])
+ assert main.schema_manager.latest().id == 0
+ assert catalog.list_branches(main.identifier) == ['dev']
+
+
[email protected]('rest', [False, True], ids=['filesystem', 'rest'])
[email protected]('name', [' dev ', ' main ', ' 123 ', 'dev+plus'])
+def test_native_branch_names_preserve_whitespace(tmp_path,
native_rest_catalog, rest, name):
+ catalog, main = _table(tmp_path, native_rest_catalog, rest)
+ seed = {'id': 1, 'value': 10, 'p': 'a'}
+ _write(main, [seed])
+ dev = _branch(catalog, main)
+ normal = {'id': 2, 'value': 20, 'p': 'a'}
+ _write(dev, [normal])
+ catalog.create_branch(main.identifier, name, tag_name='base')
+ branch = catalog.get_table(Identifier('default', 't', branch=name))
+ assert branch.current_branch() == name
+ changed = {'id': 3, 'value': 30, 'p': 'a'}
+ _write(branch, [changed])
+ _assert_rows(branch, [seed, changed])
+ _assert_rows(dev, [seed, normal])
+ _assert_rows(main, [seed])
+ assert set(catalog.list_branches(main.identifier)) == {'dev', name}
+ catalog.drop_branch(main.identifier, name)
+ _assert_rows(dev, [seed, normal])
+ _assert_rows(main, [seed])
+ assert catalog.list_branches(main.identifier) == ['dev']
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 4757c670f0..85b72f3ec7 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -22,7 +22,6 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.catalog.catalog_environment import CatalogEnvironment
-from pypaimon.common.identifier import Identifier
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.options.options import Options
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
@@ -530,7 +529,7 @@ def test_disabled_option_never_initializes_native(tmp_path):
commit.close()
[email protected]('kind', ['custom-env', 'branch', 'custom-io'])
[email protected]('kind', ['custom-env', 'custom-io'])
def test_incompatible_publication_environment_is_not_reconstructed(
tmp_path, native_rest_catalog, kind):
table = _table(tmp_path, catalog=native_rest_catalog)
@@ -538,8 +537,6 @@ def
test_incompatible_publication_environment_is_not_reconstructed(
class CustomEnvironment(CatalogEnvironment):
pass
table.catalog_environment = CustomEnvironment()
- elif kind == 'branch':
- table.identifier = Identifier('default', 't', branch='dev')
else:
table.file_io = Mock()
with patch('pypaimon.write.native_commit.native_commit_available',
return_value=True), \
diff --git a/paimon-python/pypaimon/tests/rest/api_test.py
b/paimon-python/pypaimon/tests/rest/api_test.py
index 94ab827cbf..a09a3b9c89 100644
--- a/paimon-python/pypaimon/tests/rest/api_test.py
+++ b/paimon-python/pypaimon/tests/rest/api_test.py
@@ -38,6 +38,26 @@ from pypaimon.tests.rest.rest_server import RESTCatalogServer
class ApiTest(unittest.TestCase):
+ def test_resource_names_match_java_form_url_encoding(self):
+ from pypaimon.api.rest_util import RESTUtil
+ from pypaimon.api.resource_paths import ResourcePaths
+
+ # Golden values from Java URLEncoder / URLDecoder, not from Python's
+ # encoder. Spaces and literal '+' characters must remain distinct.
+ cases = [
+ ("a b", "a+b"),
+ ("a+b", "a%2Bb"),
+ ("t$branch_ dev ", "t%24branch_+dev+"),
+ ("*~/.?&=", "*%7E%2F.%3F%26%3D"),
+ ("分支", "%E5%88%86%E6%94%AF"),
+ ]
+ for value, encoded in cases:
+ with self.subTest(value=value):
+ self.assertEqual(RESTUtil.encode_string(value), encoded)
+ self.assertEqual(RESTUtil.decode_string(encoded), value)
+ self.assertEqual(ResourcePaths("test").table("db", value),
+ "/v1/test/databases/db/tables/" + encoded)
+
def test_parse_data(self):
simple_type_test_cases = [
"DECIMAL",
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index 1c5826b03f..b7769eb22f 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -378,8 +378,12 @@ class RESTCatalogServer:
rename_request = JSON.from_json(data, RenameTableRequest)
source_table = rename_request.source
destination_table = rename_request.destination
- source =
self.table_metadata_store.get(source_table.get_full_name())
-
self.table_metadata_store.update({destination_table.get_full_name(): source})
+ source_name = source_table.get_full_name()
+ destination_name = destination_table.get_full_name()
+ if source_name not in self.table_metadata_store:
+ raise TableNotExistException(source_table)
+ if destination_name in self.table_metadata_store:
+ raise TableAlreadyExistException(destination_table)
source_table_dir = (Path(self.data_path) / self.warehouse
/ source_table.get_database_name() /
source_table.get_object_name())
destination_table_dir = (Path(self.data_path) / self.warehouse
@@ -387,7 +391,12 @@ class RESTCatalogServer:
if not source_table_dir.exists():
destination_table_dir.mkdir(parents=True)
else:
+ destination_table_dir.parent.mkdir(parents=True,
exist_ok=True)
source_table_dir.rename(destination_table_dir)
+ for store in (self.table_metadata_store, self.branch_store,
self.tag_store,
+ self.table_latest_snapshot_store,
self.table_partitions_store):
+ if source_name in store:
+ store[destination_name] = store.pop(source_name)
return self._mock_response("", 200)
# Global functions endpoint (catalog-scoped)
@@ -512,6 +521,9 @@ class RESTCatalogServer:
ErrorResponse.RESOURCE_TYPE_DEFINITION, e.name, str(e), 404
)
return self._mock_response(response, 404)
+ except ValueError as e:
+ response = ErrorResponse(None, None, str(e), 400)
+ return self._mock_response(response, 400)
except Exception as e:
self.logger.error(f"Unexpected error: {e}")
response = ErrorResponse(None, None, str(e), 500)
@@ -876,6 +888,11 @@ class RESTCatalogServer:
# ======================= Branch Handlers ================================
+ @staticmethod
+ def _validate_branch_name(branch_name):
+ from pypaimon.branch.branch_manager import BranchManager
+ BranchManager.validate_branch(branch_name)
+
def _branches_handle(self, method: str, data: str,
identifier: Identifier) -> Tuple[str, int]:
"""Handle the table-scoped branches collection (POST create / GET
list)."""
@@ -884,18 +901,26 @@ class RESTCatalogServer:
if method == "POST":
request = JSON.from_json(data, CreateBranchRequest)
+ self._validate_branch_name(request.branch)
store = self.branch_store.setdefault(identifier.get_full_name(),
set())
if request.branch in store:
raise BranchAlreadyExistException(request.branch)
+ schema_id =
self.table_metadata_store[identifier.get_full_name()].schema.id
if request.from_tag is not None:
tags = self.tag_store.get(identifier.get_full_name(), {})
if request.from_tag not in tags:
raise TagNotExistException(request.from_tag)
snapshot = tags[request.from_tag].snapshot
if snapshot is not None:
+ schema_id = snapshot.schema_id
self._write_snapshot_files(
identifier, snapshot, None, request.branch)
+ manager = self._schema_manager(identifier)
+ target = manager.copy_with_branch(request.branch)
+ for schema in manager.list_all():
+ if schema.id <= schema_id:
+ target.commit(schema)
store.add(request.branch)
return self._mock_response("", 200)
@@ -914,8 +939,11 @@ class RESTCatalogServer:
store = self.branch_store.get(identifier.get_full_name(), set())
if method == "DELETE":
+ self._validate_branch_name(branch_name)
if branch_name not in store:
raise BranchNotExistException(branch_name)
+ manager = self._schema_manager(identifier, branch_name)
+ manager.file_io.delete(manager.branch_path, recursive=True)
store.discard(branch_name)
return self._mock_response("", 200)
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
@@ -931,8 +959,13 @@ class RESTCatalogServer:
if from_branch not in store:
raise BranchNotExistException(from_branch)
request = JSON.from_json(data, RenameBranchRequest)
+ self._validate_branch_name(from_branch)
+ self._validate_branch_name(request.to_branch)
if request.to_branch in store:
raise BranchAlreadyExistException(request.to_branch)
+ source = self._schema_manager(identifier, from_branch)
+ target = self._schema_manager(identifier, request.to_branch)
+ Path(source.branch_path).rename(target.branch_path)
store.discard(from_branch)
store.add(request.to_branch)
return self._mock_response("", 200)
@@ -1035,7 +1068,8 @@ class RESTCatalogServer:
if method == "GET":
if identifier.get_full_name() not in self.table_metadata_store:
raise TableNotExistException(identifier)
- table_metadata =
self.table_metadata_store[identifier.get_full_name()]
+ table_metadata = self._branch_table_metadata(
+ identifier, (response_identifier or
identifier).get_branch_name())
table_path = (f'file://{self.data_path}/{self.warehouse}/'
f'{identifier.get_database_name()}/{identifier.get_object_name()}')
schema = table_metadata.schema.to_schema()
@@ -1045,7 +1079,7 @@ class RESTCatalogServer:
elif method == "POST":
# Alter table
request_body = JSON.from_json(data, AlterTableRequest)
- self._alter_table_impl(identifier, request_body.changes)
+ self._alter_table_impl(response_identifier or identifier,
request_body.changes)
return self._mock_response("", 200)
elif method == "DELETE":
# Drop table
@@ -1296,10 +1330,7 @@ class RESTCatalogServer:
from pypaimon.common.options.options import Options
from pypaimon.table.file_store_table import FileStoreTable
- table_metadata =
self.table_metadata_store.get(identifier.get_full_name())
- if table_metadata is None:
- raise TableNotExistException(identifier)
-
+ table_metadata = self._branch_table_metadata(identifier, branch)
table_schema = table_metadata.schema
table_path = (
f'file://{self.data_path}/{self.warehouse}/'
@@ -1457,9 +1488,28 @@ class RESTCatalogServer:
return '^' + ''.join(regex) + '$'
+ def _schema_manager(self, identifier: Identifier, branch: str = None):
+ table_path = (Path(self.data_path) / self.warehouse /
+ identifier.get_database_name() /
identifier.get_table_name())
+ return SchemaManager(self._get_file_io(), str(table_path), branch or
'main')
+
+ def _branch_table_metadata(self, identifier: Identifier, branch: str =
None):
+ base = Identifier.create(identifier.get_database_name(),
identifier.get_table_name())
+ metadata = self.table_metadata_store.get(base.get_full_name())
+ if metadata is None:
+ raise TableNotExistException(base)
+ if branch is None or branch == 'main':
+ return metadata
+ if branch not in self.branch_store.get(base.get_full_name(), set()):
+ raise BranchNotExistException(branch)
+ schema = self._schema_manager(base, branch).latest()
+ if schema is None:
+ raise BranchNotExistException(branch)
+ return TableMetadata(schema=schema, uuid=metadata.uuid,
is_external=metadata.is_external)
+
def _alter_table_impl(self, identifier: Identifier, changes: List) -> None:
- if identifier.get_full_name() not in self.table_metadata_store:
- raise TableNotExistException(identifier)
+ branch = identifier.get_branch_name()
+ table_metadata = self._branch_table_metadata(identifier, branch)
schema_changes = []
for change in changes:
@@ -1471,13 +1521,7 @@ class RESTCatalogServer:
else:
schema_changes.append(change)
- table_metadata = self.table_metadata_store[identifier.get_full_name()]
-
- table_path = (
- Path(self.data_path) / self.warehouse /
- identifier.get_database_name() / identifier.get_object_name()
- )
- schema_manager = SchemaManager(self._get_file_io(), str(table_path))
+ schema_manager = self._schema_manager(identifier, branch)
new_schema = schema_manager.commit_changes(schema_changes)
updated_metadata = TableMetadata(
@@ -1485,7 +1529,9 @@ class RESTCatalogServer:
is_external=table_metadata.is_external,
uuid=table_metadata.uuid
)
- self.table_metadata_store[identifier.get_full_name()] =
updated_metadata
+ if branch is None or branch == 'main':
+ base = Identifier.create(identifier.get_database_name(),
identifier.get_table_name())
+ self.table_metadata_store[base.get_full_name()] = updated_metadata
def _get_file_io(self):
"""Get FileIO instance for SchemaManager"""
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 12f6e36cb5..6a12343bbd 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -113,9 +113,10 @@ def _native_rest_table(table, schema_json):
catalog_options = _catalog_context_options(table)
catalog_options['metastore'] = 'rest'
native_table = NativeCatalog(catalog_options).get_table((
- table.identifier.get_database_name(),
table.identifier.get_table_name()))
+ table.identifier.get_database_name(),
table.identifier.get_object_name()))
if (native_table.location() != table.table_path
- or native_table.rest_table_uuid() !=
table.catalog_environment.uuid):
+ or native_table.rest_table_uuid() != table.catalog_environment.uuid
+ or native_table.branch() != table.current_branch()):
return None
return native_table.copy_with_resolved_schema(schema_json)
@@ -128,11 +129,9 @@ def create_native_write_table(table):
from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
from pypaimon.table.file_store_table import FileStoreTable
- # Native writes currently require the main branch.
environment = table.catalog_environment
if (type(table) is not FileStoreTable
or type(environment) is not CatalogEnvironment
- or table.current_branch() != 'main'
or table.options.query_auth_enabled):
return None
@@ -161,7 +160,7 @@ def create_native_write_table(table):
return NativeTable.from_resolved_schema(
table.table_path, schema_json,
database=table.identifier.get_database_name(),
- table=table.identifier.get_table_name(),
+ table=table.identifier.get_table_name(), branch=table.current_branch(),
options=file_io_options)
