[core] Introduce Catalog#loadTableMetadata

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -668,7 +668,8 @@ public Path newDatabasePath(String database) {
         return newDatabasePath(warehouse(), database);
     }
 
-    protected TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
+    @Override
+    public TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
         return new TableMetadata(loadTableSchema(identifier), false, null);
     }
 

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -166,6 +166,15 @@ void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNo
      */
     Table getTableById(String tableId) throws TableIdNotExistException;
 
+    /**
+     * Return a {@link TableMetadata} identified by the given {@link Identifier}.
+     *
+     * @param identifier Path of the table
+     * @return The requested table metadata
+     * @throws TableNotExistException if the target does not exist
+     */
+    TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException;
+
     /**
      * Get names of all tables under this database. An empty list is returned if none exists.
      *

paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java

@@ -158,6 +158,11 @@ public Table getTableById(String tableId) throws TableIdNotExistException {
         return wrapped.getTableById(tableId);
     }
 
+    @Override
+    public TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
+        return wrapped.loadTableMetadata(identifier);
+    }
+
     @Override
     public void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotExists)
             throws TableNotExistException, TableAlreadyExistException {

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -453,7 +453,8 @@ public void rollbackSchema(Identifier identifier, long schemaId)
         }
     }
 
-    private TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
+    @Override
+    public TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
         // if the table is system table, we need to load table metadata from the system table's data
         // table
         Identifier loadTableIdentifier =

paimon-core/src/main/java/org/apache/paimon/table/FileStoreTableFactory.java

@@ -19,7 +19,9 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
@@ -192,17 +194,31 @@ private static FileStoreTable createOtherBranchTable(
             CatalogEnvironment catalogEnvironment) {
         Options branchOptions = new Options(dynamicOptions.toMap());
         branchOptions.set(CoreOptions.BRANCH, branchName);
-        Optional<TableSchema> schema = new SchemaManager(fileIO, tablePath, branchName).latest();
+
+        Identifier identifier = catalogEnvironment.identifier();
+        Identifier branchIdentifier = null;
+        if (identifier != null) {
+            branchIdentifier =
+                    new Identifier(
+                            identifier.getDatabaseName(), identifier.getObjectName(), branchName);
+        }
+        CatalogLoader catalogLoader = catalogEnvironment.catalogLoader();
+
+        Optional<TableSchema> schema;
+        if (branchIdentifier != null && catalogLoader != null) {
+            try (Catalog catalog = catalogLoader.load()) {
+                schema = Optional.of(catalog.loadTableMetadata(branchIdentifier).schema());
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        } else {
+            schema = new SchemaManager(fileIO, tablePath, branchName).latest();
+        }
+
         if (schema.isPresent()) {
-            Identifier identifier = catalogEnvironment.identifier();
             CatalogEnvironment branchCatalogEnvironment = catalogEnvironment;
-            if (identifier != null) {
-                branchCatalogEnvironment =
-                        catalogEnvironment.copy(
-                                new Identifier(
-                                        identifier.getDatabaseName(),
-                                        identifier.getObjectName(),
-                                        branchName));
+            if (branchIdentifier != null) {
+                branchCatalogEnvironment = catalogEnvironment.copy(branchIdentifier);
             }
             return createWithoutFallbackBranch(
                     fileIO, tablePath, schema.get(), branchOptions, branchCatalogEnvironment);

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -770,7 +770,7 @@ public org.apache.paimon.table.Table getTable(Identifier identifier)
     }
 
     @Override
-    protected TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
+    public TableMetadata loadTableMetadata(Identifier identifier) throws TableNotExistException {
         return loadTableMetadata(identifier, getHmsTable(identifier));
     }