From 1ac417fbbbe9f65a13526d76cf7d14d071527252 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Tue, 27 Oct 2026 08:17:01 -0700 Subject: [PATCH] [python] Load system tables and reject branches in JDBC get_table JdbcCatalog.get_table ignored the $ suffix of the identifier: for example db.t$snapshots, db.t$options and db.t$nosuch all returned the data table, and db.t$branch_dev returned a table with main's schema even when that branch did not exist. As in Java, system tables are now loaded through the shared system table loader, as in the filesystem catalog, and unknown names raise TableNotExistException. As in Java JdbcCatalog, a branch other than main raises NotImplementedError; as in alter_table, only a branch in the name counts, so Identifier('db', 't', branch='MAIN') still reads db.t. --- .../pypaimon/catalog/jdbc_catalog.py | 22 ++++++- .../tests/catalog_table_guard_test.py | 7 ++- .../pypaimon/tests/jdbc_catalog_test.py | 62 +++++++++++++++++++ 3 files changed, 86 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/catalog/jdbc_catalog.py b/paimon-python/pypaimon/catalog/jdbc_catalog.py index 6ada44de64dc..d4dd2d237300 100644 --- a/paimon-python/pypaimon/catalog/jdbc_catalog.py +++ b/paimon-python/pypaimon/catalog/jdbc_catalog.py @@ -404,20 +404,38 @@ def list_tables(self, database_name: str) -> List[str]: def get_table(self, identifier: Union[str, Identifier]) -> Table: if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) + branch = encoded_branch_name(identifier) + if branch is not None and branch != DEFAULT_MAIN_BRANCH: + raise NotImplementedError("JdbcCatalog currently does not support table branches") if self.catalog_options.contains(CoreOptions.SCAN_FALLBACK_BRANCH): raise ValueError(f"Unsupported CoreOption {CoreOptions.SCAN_FALLBACK_BRANCH}") if not self._table_exists(identifier): raise TableNotExistException(identifier) table_path = self.get_table_path(identifier) table_schema = self.get_table_schema(identifier) + table_identifier = identifier + if identifier.is_system_table(): + table_identifier = Identifier.create( + identifier.get_database_name(), + identifier.get_table_name(), + branch=branch, + ) from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader catalog_environment = CatalogEnvironment( - identifier=identifier, + identifier=table_identifier, uuid=None, catalog_loader=JdbcCatalogLoader(self.context), supports_version_management=False ) - return FileStoreTable(self.file_io, identifier, table_path, table_schema, catalog_environment) + table = FileStoreTable(self.file_io, table_identifier, table_path, table_schema, catalog_environment) + if identifier.is_system_table(): + from pypaimon.table.system import system_table_loader + + sys_table = system_table_loader.load(identifier.get_system_table_name().lower(), table) + if sys_table is None: + raise TableNotExistException(identifier) + return sys_table + return table def create_table(self, identifier: Union[str, Identifier], schema: Schema, ignore_if_exists: bool) -> None: from pypaimon.common.options.options_utils import OptionsUtils diff --git a/paimon-python/pypaimon/tests/catalog_table_guard_test.py b/paimon-python/pypaimon/tests/catalog_table_guard_test.py index bf7ed0c8521b..1665143e9465 100644 --- a/paimon-python/pypaimon/tests/catalog_table_guard_test.py +++ b/paimon-python/pypaimon/tests/catalog_table_guard_test.py @@ -31,7 +31,7 @@ from pypaimon.schema.schema_change import SchemaChange -def assert_table_guards(test, catalog, database): +def assert_table_guards(test, catalog, database, branch_reads=True): """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) @@ -72,7 +72,8 @@ def assert_table_guards(test, catalog, database): 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')) + if branch_reads: + 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) @@ -120,7 +121,7 @@ def test_jdbc_catalog(self): "uri": "jdbc:sqlite:" + os.path.join(self.temp_dir, "catalog.db"), "catalog-key": "test-jdbc-catalog", }) - assert_table_guards(self, catalog, 'db') + assert_table_guards(self, catalog, 'db', branch_reads=False) 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'): diff --git a/paimon-python/pypaimon/tests/jdbc_catalog_test.py b/paimon-python/pypaimon/tests/jdbc_catalog_test.py index 9b017bacffec..2572e7a0ade2 100644 --- a/paimon-python/pypaimon/tests/jdbc_catalog_test.py +++ b/paimon-python/pypaimon/tests/jdbc_catalog_test.py @@ -33,8 +33,10 @@ ) from pypaimon.catalog.jdbc_catalog import JdbcCatalog, _convert_qmark_placeholders from pypaimon.catalog.rest.property_change import PropertyChange +from pypaimon.common.identifier import Identifier from pypaimon.schema.data_types import AtomicType, DataField from pypaimon.schema.schema_change import SchemaChange +from pypaimon.table.system.options_table import OptionsTable class JdbcCatalogTest(unittest.TestCase): @@ -171,6 +173,66 @@ def test_table(self): with self.assertRaises(TableNotExistException): reloaded.get_table("test_db.renamed_table") + def test_get_system_table(self): + fields = [DataField.from_dict({"id": 1, "name": "f0", "type": "INT"})] + catalog = CatalogFactory.create(self.options) + catalog.create_database("test_db", False) + catalog.create_table( + "test_db.test_table", Schema(fields=fields, options={"bucket": "1"}), False + ) + + for suffix in ("$options", "$OPTIONS", "$branch_main$options"): + name = "test_db.test_table" + suffix + for identifier in (name, Identifier.from_string(name)): + with self.subTest(identifier=identifier): + table = catalog.get_table(identifier) + self.assertIsInstance(table, OptionsTable) + self.assertEqual(table.base_table.identifier, Identifier("test_db", "test_table")) + read_builder = table.new_read_builder() + rows = read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits()) + self.assertIn({"key": "bucket", "value": "1"}, rows.to_pylist()) + + def test_get_system_table_not_exist(self): + fields = [DataField.from_dict({"id": 1, "name": "f0", "type": "INT"})] + catalog = CatalogFactory.create(self.options) + catalog.create_database("test_db", False) + catalog.create_table("test_db.test_table", Schema(fields=fields), False) + + for name in ("test_db.test_table$nosuch", "test_db.missing$options"): + with self.subTest(name=name): + with self.assertRaises(TableNotExistException) as context: + catalog.get_table(name) + self.assertEqual(str(context.exception), "Table {} does not exist".format(name)) + + def test_get_table_branch(self): + fields = [DataField.from_dict({"id": 1, "name": "f0", "type": "INT"})] + catalog = CatalogFactory.create(self.options) + catalog.create_database("test_db", False) + catalog.create_table("test_db.test_table", Schema(fields=fields), False) + + for identifier in ( + "test_db.test_table", + "test_db.test_table$branch_main", + Identifier("test_db", "test_table", branch="main"), + Identifier("test_db", "test_table", branch="MAIN"), + ): + with self.subTest(identifier=identifier): + self.assertEqual(catalog.get_table(identifier).fields, fields) + + for identifier in ( + "test_db.test_table$branch_dev", + "test_db.test_table$branch_dev$options", + "test_db.test_table$branch_dev$nosuch", + "test_db.missing$branch_dev", + "test_db.test_table$branch_MAIN", + "test_db.test_table$branch_", + ): + with self.subTest(identifier=identifier): + with self.assertRaisesRegex( + NotImplementedError, "^JdbcCatalog currently does not support table branches$" + ): + catalog.get_table(identifier) + def test_create_table_rolls_back_metadata_on_failure(self): fields = [DataField.from_dict({"id": 1, "name": "f0", "type": "INT"})] catalog = CatalogFactory.create(self.options)