[core] Fix main table's CatalogEnvironment used when switching branch on fallback table

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

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
@@ -154,12 +155,24 @@ protected FileStoreTable switchWrappedToBranch(String branchName) {
         Options branchOptions = new Options(branchSchema.options());
         branchOptions.set(CoreOptions.BRANCH, branchName);
         branchSchema = branchSchema.copy(branchOptions.toMap());
+
+        // Create branch-aware CatalogEnvironment so that REST snapshot loading
+        // targets the branch table (e.g. tableName$branch_branchName) instead of main.
+        CatalogEnvironment wrappedEnv = wrapped.catalogEnvironment();
+        CatalogEnvironment branchEnv = wrappedEnv;
+        Identifier wrappedId = wrappedEnv.identifier();
+        if (wrappedId != null) {
+            branchEnv =
+                    wrappedEnv.copy(
+                            new Identifier(
+                                    wrappedId.getDatabaseName(),
+                                    wrappedId.getTableName(),
+                                    branchName,
+                                    wrappedId.getSystemTableName()));
+        }
+
         return FileStoreTableFactory.createWithoutFallbackBranch(
-                wrapped.fileIO(),
-                wrapped.location(),
-                branchSchema,
-                new Options(),
-                wrapped.catalogEnvironment());
+                wrapped.fileIO(), wrapped.location(), branchSchema, new Options(), branchEnv);
     }
 
     protected Map<String, String> rewriteOtherOptions(Map<String, String> options) {

paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileIOFinder;
@@ -332,6 +333,44 @@ public org.apache.paimon.reader.RecordReader<InternalRow> createReader(Split spl
         };
     }
 
+    @Test
+    void testSwitchToBranch() throws Exception {
+        String branchName = "bc";
+
+        Identifier mainId = Identifier.create("mydb", "mytable");
+        CatalogEnvironment env =
+                new CatalogEnvironment(mainId, "uuid-1", null, null, null, null, false, false);
+
+        TableSchema tableSchema =
+                SchemaUtils.forceCommit(
+                        new SchemaManager(LocalFileIO.create(), tablePath),
+                        new Schema(
+                                ROW_TYPE.getFields(),
+                                Collections.singletonList("pt"),
+                                Collections.emptyList(),
+                                Collections.emptyMap(),
+                                ""));
+        AppendOnlyFileStoreTable mainTable =
+                new AppendOnlyFileStoreTable(fileIO, tablePath, tableSchema, env);
+
+        writeDataIntoTable(mainTable, 0, rowData(1, 10));
+        mainTable.createBranch(branchName);
+
+        FileStoreTable branchTable = createTableFromBranch(mainTable, branchName);
+        writeDataIntoTable(branchTable, 0, rowData(2, 20));
+
+        FallbackReadFileStoreTable fallbackTable =
+                new FallbackReadFileStoreTable(mainTable, branchTable, true);
+
+        FileStoreTable switched = fallbackTable.switchToBranch(branchName);
+        Identifier switchedId = switched.catalogEnvironment().identifier();
+
+        assertThat(switchedId).isNotNull();
+        assertThat(switchedId.getDatabaseName()).isEqualTo("mydb");
+        assertThat(switchedId.getBranchName()).isEqualTo(branchName);
+        assertThat(switchedId.getObjectName()).isEqualTo("mytable$branch_bc");
+    }
+
     private void writeDataIntoTable(
             FileStoreTable table, long commitIdentifier, InternalRow... allData) throws Exception {
         StreamTableWrite write = table.newWrite(commitUser);