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 ba68b107a7 [python] Reject branch and system tables in catalog table
operations (#10822)
ba68b107a7 is described below
commit ba68b107a7be5137ec1ce99fddf4889bf934bcd1
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Oct 27 14:16:14 2026 +0800
[python] Reject branch and system tables in catalog table operations
(#10822)
---
paimon-python/pypaimon/catalog/catalog_utils.py | 33 ++++++
.../pypaimon/catalog/filesystem_catalog.py | 10 ++
paimon-python/pypaimon/catalog/jdbc_catalog.py | 20 +++-
.../pypaimon/catalog/rest/rest_catalog.py | 10 ++
.../pypaimon/tests/catalog_table_guard_test.py | 115 +++++++++++++++++++++
.../pypaimon/tests/rest/rest_catalog_test.py | 4 +
.../tests/rest/rest_permission_server_test.py | 4 +-
7 files changed, 193 insertions(+), 3 deletions(-)
diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py
b/paimon-python/pypaimon/catalog/catalog_utils.py
index a3f70ab5b7..492d60da9d 100644
--- a/paimon-python/pypaimon/catalog/catalog_utils.py
+++ b/paimon-python/pypaimon/catalog/catalog_utils.py
@@ -21,9 +21,42 @@ import re
from typing import Optional
from pypaimon.api.api_response import PagedList, Partition
+from pypaimon.common.identifier import Identifier
from pypaimon.table.table import Table
+def encoded_branch_name(identifier: Identifier) -> Optional[str]:
+ """The branch in the encoded name of ``identifier``, as in
``t$branch_dev``, or None.
+
+ ``Identifier(db, t, branch='main')``, in any case, is encoded as and equal
to
+ ``db.t``, so it has none. ``t$branch_MAIN`` names a branch ``MAIN``.
+ """
+ without_branch = Identifier(
+ identifier.get_database_name(), identifier.get_table_name(),
+ system_table=identifier.get_system_table_name())
+ return None if identifier == without_branch else
identifier.get_branch_name()
+
+
+def check_not_branch(identifier: Identifier, method: str) -> None:
+ """Mirrors Java ``CatalogUtils.checkNotBranch``.
+
+ Unlike Java, ``Identifier(db, t, branch='main')`` is accepted, see
+ ``encoded_branch_name``.
+ """
+ if encoded_branch_name(identifier) is not None:
+ raise ValueError(
+ "Cannot '{}' for branch table '{}', please modify the table with
the default branch."
+ .format(method, identifier.get_full_name()))
+
+
+def check_not_system_table(identifier: Identifier, method: str) -> None:
+ """Mirrors Java ``CatalogUtils.checkNotSystemTable``."""
+ if identifier.is_system_table():
+ raise ValueError(
+ "Cannot '{}' for system table '{}', please use data table."
+ .format(method, identifier.get_full_name()))
+
+
def list_partitions_from_file_system(
table: Table,
max_results: Optional[int] = None,
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index 5e169021ca..76218a8bf8 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -21,6 +21,7 @@ from pypaimon.api.api_response import GetTagResponse,
PagedList, Partition
from pypaimon.catalog.catalog import Catalog
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_environment import CatalogEnvironment
+from pypaimon.catalog.catalog_utils import check_not_branch,
check_not_system_table
from pypaimon.catalog.catalog_exception import (BranchAlreadyExistException,
BranchNotExistException,
DatabaseAlreadyExistException,
@@ -189,6 +190,8 @@ class FileSystemCatalog(Catalog):
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "create_table")
+ check_not_system_table(identifier, "create_table")
self.get_database(identifier.get_database_name())
try:
self.get_table(identifier)
@@ -229,6 +232,7 @@ class FileSystemCatalog(Catalog):
) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_system_table(identifier, "alter_table")
try:
self.get_table(identifier)
except TableNotExistException:
@@ -253,6 +257,10 @@ class FileSystemCatalog(Catalog):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
target_identifier = Identifier.from_string(target_identifier)
+ check_not_branch(source_identifier, "rename_table")
+ check_not_branch(target_identifier, "rename_table")
+ check_not_system_table(source_identifier, "rename_table")
+ check_not_system_table(target_identifier, "rename_table")
# Verify source table exists
try:
@@ -277,6 +285,8 @@ class FileSystemCatalog(Catalog):
def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "drop_table")
+ check_not_system_table(identifier, "drop_table")
# Check if table exists
try:
diff --git a/paimon-python/pypaimon/catalog/jdbc_catalog.py
b/paimon-python/pypaimon/catalog/jdbc_catalog.py
index da267fbdbf..770a9c5f24 100644
--- a/paimon-python/pypaimon/catalog/jdbc_catalog.py
+++ b/paimon-python/pypaimon/catalog/jdbc_catalog.py
@@ -24,6 +24,11 @@ from urllib.parse import parse_qs, urlparse
from pypaimon.catalog.catalog import Catalog
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_environment import CatalogEnvironment
+from pypaimon.catalog.catalog_utils import (
+ check_not_branch,
+ check_not_system_table,
+ encoded_branch_name
+)
from pypaimon.catalog.catalog_exception import (
DatabaseAlreadyExistException,
DatabaseNotExistException,
@@ -33,7 +38,7 @@ from pypaimon.catalog.catalog_exception import (
from pypaimon.catalog.database import Database
from pypaimon.catalog.rest.property_change import PropertyChange
from pypaimon.common.file_io import FileIO
-from pypaimon.common.identifier import Identifier
+from pypaimon.common.identifier import DEFAULT_MAIN_BRANCH, Identifier
from pypaimon.common.options.config import CatalogOptions, JdbcCatalogOptions
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.schema.schema import Schema
@@ -417,6 +422,8 @@ class JdbcCatalog(Catalog):
raise ValueError(f"The value of {CoreOptions.AUTO_CREATE.key()}
property should be False.")
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "create_table")
+ check_not_system_table(identifier, "create_table")
self.get_database(identifier.get_database_name())
if self._table_exists(identifier):
if not ignore_if_exists:
@@ -444,6 +451,8 @@ class JdbcCatalog(Catalog):
def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "drop_table")
+ check_not_system_table(identifier, "drop_table")
if not self._table_exists(identifier):
if not ignore_if_not_exists:
raise TableNotExistException(identifier)
@@ -466,6 +475,10 @@ class JdbcCatalog(Catalog):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
target_identifier = Identifier.from_string(target_identifier)
+ check_not_branch(source_identifier, "rename_table")
+ check_not_branch(target_identifier, "rename_table")
+ check_not_system_table(source_identifier, "rename_table")
+ check_not_system_table(target_identifier, "rename_table")
if not self._table_exists(source_identifier):
raise TableNotExistException(source_identifier)
self.get_database(target_identifier.get_database_name())
@@ -515,10 +528,15 @@ class JdbcCatalog(Catalog):
) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_system_table(identifier, "alter_table")
if not self._table_exists(identifier):
if not ignore_if_not_exists:
raise TableNotExistException(identifier)
return
+ # As in Java JdbcCatalog#alterTableImpl, only the main branch can be
altered.
+ branch = encoded_branch_name(identifier)
+ if branch is not None and branch != DEFAULT_MAIN_BRANCH:
+ raise NotImplementedError("JdbcCatalog currently does not support
table branches")
schema_manager = SchemaManager(self.file_io,
self.get_table_path(identifier))
table_schema = schema_manager.commit_changes(changes)
if self._sync_all_properties():
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index c561a1d03f..2289eaa174 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -28,6 +28,7 @@ from pypaimon.api.rest_exception import
(NoSuchResourceException, AlreadyExistsE
from pypaimon.catalog.catalog import Catalog
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_environment import CatalogEnvironment
+from pypaimon.catalog.catalog_utils import check_not_branch,
check_not_system_table
from pypaimon.catalog.catalog_exception import (
ColumnAlreadyExistException, ColumnNotExistException,
TableNotExistException, DatabaseAlreadyExistException,
@@ -295,6 +296,8 @@ class RESTCatalog(Catalog):
def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "create_table")
+ check_not_system_table(identifier, "create_table")
try:
self.rest_api.create_table(identifier, schema)
except AlreadyExistsException as e:
@@ -309,6 +312,10 @@ class RESTCatalog(Catalog):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
target_identifier = Identifier.from_string(target_identifier)
+ check_not_branch(source_identifier, "rename_table")
+ check_not_branch(target_identifier, "rename_table")
+ check_not_system_table(source_identifier, "rename_table")
+ check_not_system_table(target_identifier, "rename_table")
try:
self.rest_api.rename_table(source_identifier, target_identifier)
except NoSuchResourceException as e:
@@ -321,6 +328,8 @@ class RESTCatalog(Catalog):
def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_branch(identifier, "drop_table")
+ check_not_system_table(identifier, "drop_table")
try:
self.rest_api.drop_table(identifier)
except NoSuchResourceException as e:
@@ -423,6 +432,7 @@ class RESTCatalog(Catalog):
) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
+ check_not_system_table(identifier, "alter_table")
try:
self.rest_api.alter_table(identifier, changes)
except NoSuchResourceException as e:
diff --git a/paimon-python/pypaimon/tests/catalog_table_guard_test.py
b/paimon-python/pypaimon/tests/catalog_table_guard_test.py
new file mode 100644
index 0000000000..2787b04963
--- /dev/null
+++ b/paimon-python/pypaimon/tests/catalog_table_guard_test.py
@@ -0,0 +1,115 @@
+################################################################################
+# 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.
+#################################################################################
+
+import os
+import shutil
+import tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.identifier import Identifier
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
+
+
+def assert_table_guards(test, catalog, database):
+ """Table operations reject branch and system tables, as in Java, and leave
the table untouched."""
+ pa_schema = pa.schema([('id', pa.int32())])
+ schema = Schema.from_pyarrow_schema(pa_schema)
+ catalog.create_database(database, True)
+ name = database + '.t'
+ catalog.create_table(name, schema, False)
+ table = catalog.get_table(name)
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa.Table.from_pydict({'id': [1, 2]},
schema=pa_schema))
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+ table.branch_manager().create_branch('dev')
+
+ add_column = SchemaChange.add_column('x', AtomicType('INT'))
+ for target in (name + '$snapshots', name + '$branch_dev', name +
'$branch_main'):
+ operations = {
+ 'create_table': lambda: catalog.create_table(target, schema, True),
+ 'drop_table': lambda: catalog.drop_table(target, True),
+ 'rename_table': lambda: catalog.rename_table(target, database +
'.renamed'),
+ }
+ if target.endswith('$snapshots'):
+ operations['alter_table'] = lambda: catalog.alter_table(target,
[add_column])
+ for method, operation in operations.items():
+ with test.subTest(target=target, method=method):
+ with test.assertRaises(ValueError) as context:
+ operation()
+ test.assertIn("Cannot '{}'".format(method),
str(context.exception))
+ with test.subTest(target=target, method='rename_table to'):
+ with test.assertRaises(ValueError):
+ catalog.rename_table(name, target)
+
+ test.assertEqual(catalog.list_tables(database), ['t'])
+ table = catalog.get_table(name)
+ test.assertEqual([field.name for field in table.table_schema.fields],
['id'])
+ read_builder = table.new_read_builder()
+ result =
read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+ test.assertEqual(sorted(result.column('id').to_pylist()), [1, 2])
+ test.assertIsNotNone(catalog.get_table(name + '$branch_dev'))
+
+ # Identifier(db, t, branch='main') is encoded as and equal to db.t, so it
is accepted.
+ catalog.create_table(Identifier(database, 'm', branch='main'), schema,
False)
+ catalog.rename_table(Identifier(database, 'm', branch='main'),
Identifier(database, 'm2', branch='MAIN'))
+ test.assertEqual(sorted(catalog.list_tables(database)), ['m2', 't'])
+ catalog.drop_table(Identifier(database, 'm2', branch='main'), False)
+ test.assertEqual(catalog.list_tables(database), ['t'])
+
+
+class CatalogTableGuardTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="unittest_")
+ self.warehouse = os.path.join(self.temp_dir, "warehouse")
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def test_filesystem_catalog(self):
+ assert_table_guards(self, CatalogFactory.create({"warehouse":
self.warehouse}), 'db')
+
+ def test_jdbc_catalog(self):
+ catalog = CatalogFactory.create({
+ "metastore": "jdbc",
+ "warehouse": self.warehouse,
+ "uri": "jdbc:sqlite:" + os.path.join(self.temp_dir, "catalog.db"),
+ "catalog-key": "test-jdbc-catalog",
+ })
+ assert_table_guards(self, catalog, 'db')
+ add_column = SchemaChange.add_column('x', AtomicType('INT'))
+ catalog.get_table('db.t').branch_manager().create_branch('MAIN')
+ for target in ('db.t$branch_dev', 'db.t$branch_MAIN'):
+ with self.subTest(target=target):
+ with self.assertRaises(NotImplementedError):
+ catalog.alter_table(target, [add_column])
+ self.assertEqual([field.name for field in
catalog.get_table('db.t').table_schema.fields], ['id'])
+ catalog.alter_table(Identifier('db', 't', branch='MAIN'), [add_column])
+ self.assertEqual([field.name for field in
catalog.get_table('db.t').table_schema.fields], ['id', 'x'])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/rest/rest_catalog_test.py
b/paimon-python/pypaimon/tests/rest/rest_catalog_test.py
index 521be62ab5..2cf8bd163b 100644
--- a/paimon-python/pypaimon/tests/rest/rest_catalog_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_catalog_test.py
@@ -159,6 +159,10 @@ class RESTCatalogTest(RESTBaseTest):
self.assertIn("default.no_such_table", str(context.exception))
self.assertIn("does not exist", str(context.exception))
+ def test_table_operations_reject_branch_and_system_tables(self):
+ from pypaimon.tests.catalog_table_guard_test import assert_table_guards
+ assert_table_guards(self, self.rest_catalog, 'guard_db')
+
def test_table_rollback_to_snapshot(self):
"""Test table-level rollback_to_snapshot via FileStoreTable."""
table_name = "default.table_level_rollback_snapshot"
diff --git a/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
b/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
index 4d24b3a93f..71a5c0591f 100644
--- a/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
@@ -303,8 +303,8 @@ def test_a_recreated_table_starts_from_its_own_schema(rest,
drop_database):
def test_a_branch_is_not_dropped_as_a_table(rest, branch):
catalog, _ = rest
catalog.create_branch(USERS, 'b1')
- with pytest.raises(TableNotExistException):
- catalog.drop_table(Identifier('default', 'users', branch=branch))
+ with pytest.raises(NoSuchResourceException):
+ catalog.rest_api.drop_table(Identifier('default', 'users',
branch=branch))
catalog.alter_table('default.users', [SchemaChange.set_option('a', 'b')])
assert catalog.get_table('default.users').table_schema.options['a'] == 'b'
assert catalog.list_branches(USERS) == ['b1']
