[python] Load system tables and reject branches in JDBC get_table

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

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'):

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)