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)