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 69a9472a64 [python] Make schema management branch-aware (#8829)
69a9472a64 is described below
commit 69a9472a641d25ada1666ac54f01ad437184458a
Author: zhoulii <[email protected]>
AuthorDate: Thu Jul 23 21:11:10 2026 +0800
[python] Make schema management branch-aware (#8829)
---
.../pypaimon/catalog/filesystem_catalog.py | 49 +++++++++++++---------
paimon-python/pypaimon/table/file_store_table.py | 10 +++--
.../tests/filesystem_catalog_branch_test.py | 20 +++++++++
.../pypaimon/tests/rest/rest_branch_test.py | 14 +++++++
.../pypaimon/tests/table/file_store_table_test.py | 6 ++-
5 files changed, 74 insertions(+), 25 deletions(-)
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index 5c85c27ca8..76ba5e7798 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -21,23 +21,22 @@ from pypaimon.api.api_response import GetTagResponse,
PagedList
from pypaimon.catalog.catalog import Catalog
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_environment import CatalogEnvironment
-from pypaimon.catalog.catalog_exception import (
- BranchAlreadyExistException,
- BranchNotExistException,
- DatabaseAlreadyExistException,
- DatabaseNotExistException,
- TableAlreadyExistException,
- TableNotExistException,
- TagAlreadyExistException,
- TagNotExistException,
-)
+from pypaimon.catalog.catalog_exception import (BranchAlreadyExistException,
+ BranchNotExistException,
+ DatabaseAlreadyExistException,
+ DatabaseNotExistException,
+ TableAlreadyExistException,
+ TableNotExistException,
+ TagAlreadyExistException,
+ TagNotExistException)
from pypaimon.catalog.database import Database
+from pypaimon.common.file_io import FileIO
+from pypaimon.common.identifier import Identifier
from pypaimon.common.options import Options
from pypaimon.common.options.config import CatalogOptions
from pypaimon.common.options.core_options import CoreOptions
-from pypaimon.common.file_io import FileIO
-from pypaimon.common.identifier import Identifier
-from pypaimon.common.time_utils import duration_to_iso8601,
local_datetime_to_system_zone_millis
+from pypaimon.common.time_utils import (duration_to_iso8601,
+ local_datetime_to_system_zone_millis)
from pypaimon.filesystem.caching_file_io import CachingFileIO
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.schema_manager import SchemaManager
@@ -143,7 +142,8 @@ class FileSystemCatalog(Catalog):
table_schema = self.get_table_schema(identifier)
# Create catalog environment for filesystem catalog
- from pypaimon.catalog.filesystem_catalog_loader import
FileSystemCatalogLoader
+ from pypaimon.catalog.filesystem_catalog_loader import \
+ FileSystemCatalogLoader
catalog_environment = CatalogEnvironment(
identifier=identifier,
uuid=None,
@@ -190,7 +190,11 @@ class FileSystemCatalog(Catalog):
def get_table_schema(self, identifier: Identifier):
table_path = self.get_table_path(identifier)
- table_schema = SchemaManager(self.file_io, table_path).latest()
+ table_schema = SchemaManager(
+ self.file_io,
+ table_path,
+ branch=identifier.get_branch_name_or_default(),
+ ).latest()
if table_schema is None:
raise TableNotExistException(identifier)
return table_schema
@@ -219,7 +223,11 @@ class FileSystemCatalog(Catalog):
return
table_path = self.get_table_path(identifier)
- schema_manager = SchemaManager(self.file_io, table_path)
+ schema_manager = SchemaManager(
+ self.file_io,
+ table_path,
+ branch=identifier.get_branch_name_or_default(),
+ )
try:
schema_manager.commit_changes(changes)
except Exception as e:
@@ -254,7 +262,7 @@ class FileSystemCatalog(Catalog):
def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False):
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
-
+
# Check if table exists
try:
self.get_table(identifier)
@@ -262,7 +270,7 @@ class FileSystemCatalog(Catalog):
if not ignore_if_not_exists:
raise
return
-
+
# Delete the table directory
table_path = self.get_table_path(identifier)
self.file_io.delete(table_path, True)
@@ -286,9 +294,9 @@ class FileSystemCatalog(Catalog):
page_token: Optional[str] = None,
partition_name_pattern: Optional[str] = None,
):
- from pypaimon.api.api_response import Partition, PagedList
- from pypaimon.manifest.manifest_list_manager import ManifestListManager
+ from pypaimon.api.api_response import PagedList, Partition
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+ from pypaimon.manifest.manifest_list_manager import ManifestListManager
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
@@ -345,6 +353,7 @@ class FileSystemCatalog(Catalog):
# Apply pattern filter with proper regex escaping
if partition_name_pattern:
import re
+
# Escape special regex chars except '*', then replace '*' with '.*'
escaped_pattern = re.escape(partition_name_pattern).replace(r'\*',
'.*')
regex = re.compile(escaped_pattern)
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 8edbc14021..a1de56cb99 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -60,8 +60,8 @@ class FileStoreTable(Table):
self.is_primary_key_table = bool(self.primary_keys)
self.total_buckets = self.options.bucket()
- current_branch = self.options.branch()
- self.schema_manager = SchemaManager(file_io, table_path,
branch=current_branch)
+ self.schema_manager = SchemaManager(
+ file_io, table_path, branch=self.current_branch())
@classmethod
def from_path(cls, table_path: str) -> 'FileStoreTable':
@@ -118,7 +118,8 @@ class FileStoreTable(Table):
# If catalog environment has a catalog loader, use CatalogBranchManager
catalog_loader = self.catalog_environment.catalog_loader
if catalog_loader is not None and
self.catalog_environment.supports_version_management:
- from pypaimon.branch.catalog_branch_manager import
CatalogBranchManager
+ from pypaimon.branch.catalog_branch_manager import \
+ CatalogBranchManager
return CatalogBranchManager(
catalog_loader,
self.identifier
@@ -453,7 +454,8 @@ class FileStoreTable(Table):
def create_global_index(self, index_column, index_type: str = "btree",
partition_filter=None, partitions=None,
options: Optional[dict] = None) -> int:
- from pypaimon.globalindex.create_global_index import
create_global_index
+ from pypaimon.globalindex.create_global_index import \
+ create_global_index
return create_global_index(
self,
index_column,
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
index 2129f811fa..4bf22b85aa 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
@@ -36,6 +36,8 @@ from pypaimon.catalog.catalog_exception import
(BranchAlreadyExistException,
TableNotExistException,
TagNotExistException)
from pypaimon.common.identifier import Identifier
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
@@ -77,6 +79,24 @@ class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
self.catalog.create_branch(self.identifier, "b1")
self.assertEqual(self.catalog.list_branches(self.identifier), ["b1"])
+ def test_alter_table_isolated_to_branch(self):
+ self.catalog.create_branch(self.identifier, "b1")
+ branch_identifier = Identifier(
+ self.identifier.get_database_name(),
+ self.identifier.get_table_name(),
+ branch="b1",
+ )
+
+ self.catalog.alter_table(
+ branch_identifier,
+ [SchemaChange.add_column("branch_col", AtomicType("STRING"))],
+ )
+
+ self.assertNotIn(
+ "branch_col", self.catalog.get_table(self.identifier).field_names)
+ self.assertIn(
+ "branch_col",
self.catalog.get_table(branch_identifier).field_names)
+
def test_create_branch_duplicate_raises(self):
self.catalog.create_branch(self.identifier, "b1")
with self.assertRaises(BranchAlreadyExistException) as cm:
diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
index 9658c4ee3e..3b9eb80ff2 100644
--- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
@@ -51,6 +51,20 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
self.rest_catalog.create_branch(identifier, "b1")
self.assertEqual(self.rest_catalog.list_branches(identifier), ["b1"])
+ def test_branch_table_uses_branch_schema_manager(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_branch(identifier, "b1")
+ branch_identifier = Identifier(
+ identifier.get_database_name(),
+ identifier.get_table_name(),
+ branch="b1",
+ )
+
+ table = self.rest_catalog.get_table(branch_identifier)
+
+ self.assertEqual(table.current_branch(), "b1")
+ self.assertEqual(table.schema_manager.branch, "b1")
+
def test_create_branch_duplicate_raises(self):
identifier = self._identifier()
self.rest_catalog.create_branch(identifier, "b1")
diff --git a/paimon-python/pypaimon/tests/table/file_store_table_test.py
b/paimon-python/pypaimon/tests/table/file_store_table_test.py
index 2e093f4e4a..9690a550b2 100644
--- a/paimon-python/pypaimon/tests/table/file_store_table_test.py
+++ b/paimon-python/pypaimon/tests/table/file_store_table_test.py
@@ -112,7 +112,7 @@ class FileStoreTableTest(unittest.TestCase):
"""Test consumer_manager when branch is encoded in the identifier."""
branch_name = "feature_branch"
- # Create a regular table; the branch is supplied later via the
+ # Create a regular table and branch, then select the branch via the
# branch-encoded identifier (Java-aligned routing).
schema = Schema.from_pyarrow_schema(
self.pa_schema,
@@ -120,6 +120,7 @@ class FileStoreTableTest(unittest.TestCase):
options={CoreOptions.BUCKET.key(): "2"},
)
self.catalog.create_table('default.test_branch_table', schema, True)
+ self.catalog.create_branch('default.test_branch_table', branch_name)
# Access the table with a branch-encoded identifier.
branch_table = self.catalog.get_table(
@@ -267,6 +268,8 @@ class FileStoreTableTest(unittest.TestCase):
options={CoreOptions.BUCKET.key(): "2"},
)
self.catalog.create_table('default.test_changelog_branch_table',
schema, False)
+ self.catalog.create_branch(
+ 'default.test_changelog_branch_table', branch_name)
branch_table = self.catalog.get_table(
'default.test_changelog_branch_table$branch_{}'.format(branch_name))
@@ -316,6 +319,7 @@ class FileStoreTableTest(unittest.TestCase):
options={CoreOptions.BUCKET.key(): "2"},
)
self.catalog.create_table('default.test_current_branch', schema, False)
+ self.catalog.create_branch('default.test_current_branch', branch_name)
branch_table = self.catalog.get_table(
'default.test_current_branch$branch_{}'.format(branch_name))
self.assertEqual(branch_table.current_branch(), branch_name)
