diff --git a/include/paimon/catalog/table.h b/include/paimon/catalog/table.h
index ef68d1e44..59c82495f 100644
--- a/include/paimon/catalog/table.h
+++ b/include/paimon/catalog/table.h
@@ -40,7 +40,7 @@ class PAIMON_EXPORT Table {
     ///
     /// @param file_system File system used to access the table directory and schema files.
     /// @param table_path Root path of the table in the file system.
-    /// @param identifier Logical table identifier used for naming and error messages.
+    /// @param identifier Logical table identifier selecting the schema branch and table name.
     /// @return A table initialized with the latest available schema.
     static Result<std::shared_ptr<Table>> Create(const std::shared_ptr<FileSystem>& file_system,
                                                  const std::string& table_path,
diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp
index 09be18d7d..827e50b4a 100644
--- a/src/paimon/core/catalog/file_system_catalog.cpp
+++ b/src/paimon/core/catalog/file_system_catalog.cpp
@@ -105,7 +105,7 @@ Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
         return GlobalSystemTableLoader::IsSupported(identifier.GetTableName(), catalog_options_);
     }
-    // The branch component is dropped when the data table identifier is rebuilt below, so the
+    // The system table suffix is dropped when the data table identifier is rebuilt below, so the
     // identifier is validated as a whole here.
     PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
@@ -115,8 +115,9 @@ Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const
         if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) {
             return false;
         }
-        PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
-        Identifier data_identifier(identifier.GetDatabaseName(), data_table_name);
+        PAIMON_ASSIGN_OR_RAISE(std::string branch, identifier.GetBranchNameOrDefault());
+        PAIMON_ASSIGN_OR_RAISE(Identifier data_identifier,
+                               CatalogUtils::BranchIdentifier(identifier, branch));
         PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
                                TableSchemaExists(data_identifier));
         return latest_schema != std::nullopt;
@@ -181,7 +182,11 @@ Result<std::optional<std::shared_ptr<TableSchema>>> FileSystemCatalog::TableSche
             "do not support checking TableSchemaExists for system table.");
     }
     PAIMON_ASSIGN_OR_RAISE(std::string table_path, NewDataTablePath(warehouse_, identifier));
-    SchemaManager schema_manager(fs_, table_path);
+    PAIMON_ASSIGN_OR_RAISE(std::string branch, identifier.GetBranchNameOrDefault());
+    if (StringUtils::EqualsIgnoreCase(branch, Identifier::kDefaultMainBranch)) {
+        branch = Identifier::kDefaultMainBranch;
+    }
+    SchemaManager schema_manager(fs_, table_path, branch);
     return schema_manager.Latest();
 }
 
@@ -291,7 +296,7 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
                                system_table->ArrowSchema());
         return std::make_shared<SystemTableSchema>(std::move(arrow_schema));
     }
-    // The branch component is dropped when the data table identifier is rebuilt below, so the
+    // The system table suffix is dropped when the data table identifier is rebuilt below, so the
     // identifier is validated as a whole here.
     PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
@@ -301,15 +306,16 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
         if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) {
             return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
         }
-        PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
-        Identifier data_identifier(identifier.GetDatabaseName(), data_table_name);
+        PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName());
+        PAIMON_ASSIGN_OR_RAISE(
+            Identifier data_identifier,
+            CatalogUtils::BranchIdentifier(identifier, branch.value_or(std::string())));
         PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
                                TableSchemaExists(data_identifier));
         if (!latest_schema) {
             return Status::NotExist(fmt::format("{} not exist", data_identifier.ToString()));
         }
         std::map<std::string, std::string> dynamic_options;
-        PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName());
         if (branch) {
             dynamic_options[Options::BRANCH] = branch.value();
         }
diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp
index 74578e398..68aa6ac47 100644
--- a/src/paimon/core/catalog/file_system_catalog_test.cpp
+++ b/src/paimon/core/catalog/file_system_catalog_test.cpp
@@ -31,6 +31,7 @@
 #include "paimon/common/utils/path_util.h"
 #include "paimon/core/catalog/version_managed_catalog.h"
 #include "paimon/core/core_options.h"
+#include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/core/table/system/global_system_tables.h"
 #include "paimon/core/table/system/system_table_schema.h"
@@ -238,6 +239,78 @@ TEST(FileSystemCatalogTest, TestOptionsSystemTableCatalog) {
                         "Cannot 'renameTable' for system table");
 }
 
+TEST(FileSystemCatalogTest, TestBranchTableLoadsBranchSchema) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
+    ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/false));
+
+    auto typed_schema = arrow::schema(
+        {arrow::field("main_c0", arrow::int32()), arrow::field("main_c1", arrow::utf8())});
+    ::ArrowSchema c_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &c_schema).ok());
+    Identifier identifier("db1", "tbl1");
+    ASSERT_OK(catalog.CreateTable(identifier, &c_schema, {}, {}, options,
+                                  /*ignore_if_exists=*/false));
+    ArrowSchemaRelease(&c_schema);
+
+    ASSERT_OK_AND_ASSIGN(std::string table_path, catalog.GetTableLocation(identifier));
+    SchemaManager branch_schema_manager(core_options.GetFileSystem(), table_path, "b1");
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<TableSchema> branch_schema,
+        TableSchema::Create(
+            /*schema_id=*/3,
+            arrow::schema({arrow::field("b1_c0", arrow::int32(), /*nullable=*/false)}), {}, {},
+            options));
+    ASSERT_OK_AND_ASSIGN(std::string branch_schema_json, branch_schema->ToJsonString());
+    ASSERT_OK(core_options.GetFileSystem()->AtomicStore(branch_schema_manager.ToSchemaPath(3),
+                                                        branch_schema_json));
+
+    Identifier branch_identifier("db1", "tbl1$branch_b1");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> loaded_schema,
+                         catalog.LoadTableSchema(branch_identifier));
+    auto data_schema = std::dynamic_pointer_cast<DataSchema>(loaded_schema);
+    ASSERT_NE(nullptr, data_schema);
+    ASSERT_EQ((std::vector<std::string>{"b1_c0"}), data_schema->FieldNames());
+    ASSERT_EQ(3, data_schema->Id());
+
+    ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(branch_identifier));
+    ASSERT_TRUE(exists);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table, catalog.GetTable(branch_identifier));
+    data_schema = std::dynamic_pointer_cast<DataSchema>(table->LatestSchema());
+    ASSERT_NE(nullptr, data_schema);
+    ASSERT_EQ((std::vector<std::string>{"b1_c0"}), data_schema->FieldNames());
+    ASSERT_EQ(3, data_schema->Id());
+
+    for (const auto& main_identifier :
+         {Identifier("db1", "tbl1$branch_main"), Identifier("db1", "tbl1$branch_MAIN")}) {
+        ASSERT_OK_AND_ASSIGN(loaded_schema, catalog.LoadTableSchema(main_identifier));
+        data_schema = std::dynamic_pointer_cast<DataSchema>(loaded_schema);
+        ASSERT_NE(nullptr, data_schema);
+        ASSERT_EQ((std::vector<std::string>{"main_c0", "main_c1"}), data_schema->FieldNames());
+        ASSERT_EQ(0, data_schema->Id());
+    }
+
+    Identifier audit_log_identifier("db1", "tbl1$branch_b1$audit_log");
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(audit_log_identifier));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(loaded_schema, catalog.LoadTableSchema(audit_log_identifier));
+    ASSERT_EQ((std::vector<std::string>{"rowkind", "b1_c0"}), loaded_schema->FieldNames());
+
+    for (const auto& missing_identifier : {Identifier("db1", "tbl1$branch_missing"),
+                                           Identifier("db1", "tbl1$branch_missing$options")}) {
+        ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(missing_identifier));
+        ASSERT_FALSE(exists);
+        Status missing_schema = catalog.LoadTableSchema(missing_identifier).status();
+        ASSERT_TRUE(missing_schema.IsNotExist()) << missing_schema.ToString();
+    }
+}
+
 TEST(FileSystemCatalogTest, TestBranchIdentifierRejectedForTableOperations) {
     std::map<std::string, std::string> options;
     options[Options::FILE_SYSTEM] = "local";
diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp
index eaa574405..c627b4581 100644
--- a/src/paimon/core/table/table.cpp
+++ b/src/paimon/core/table/table.cpp
@@ -22,6 +22,7 @@
 
 #include "fmt/format.h"
 #include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/core/catalog/catalog_utils.h"
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
@@ -37,7 +38,12 @@ Result<std::shared_ptr<Table>> Table::Create(const std::shared_ptr<FileSystem>&
         return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
     }
 
-    SchemaManager schema_manager(file_system, table_path);
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
+    PAIMON_ASSIGN_OR_RAISE(std::string branch, identifier.GetBranchNameOrDefault());
+    if (StringUtils::EqualsIgnoreCase(branch, Identifier::kDefaultMainBranch)) {
+        branch = Identifier::kDefaultMainBranch;
+    }
+    SchemaManager schema_manager(file_system, table_path, branch);
     PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
                            schema_manager.Latest());
     if (!latest_schema) {
diff --git a/src/paimon/core/table/table_test.cpp b/src/paimon/core/table/table_test.cpp
index 60113ec8c..8032890e6 100644
--- a/src/paimon/core/table/table_test.cpp
+++ b/src/paimon/core/table/table_test.cpp
@@ -66,6 +66,36 @@ TEST(TableTest, TestCreateWithUnknownDatabase) {
     EXPECT_EQ(data_schema->PrimaryKeys(), primary_keys);
 }
 
+TEST(TableTest, TestCreateWithBranch) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fs = dir->GetFileSystem();
+    SchemaManager schema_manager(fs, dir->Str());
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<TableSchema> main_schema,
+                         schema_manager.CreateTable(
+                             arrow::schema({arrow::field("main_c0", arrow::int32())}), {}, {}, {}));
+    SchemaManager branch_schema_manager(fs, dir->Str(), "b1");
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<TableSchema> branch_schema,
+                         branch_schema_manager.CreateTable(
+                             arrow::schema({arrow::field("b1_c0", arrow::int32())}), {}, {}, {}));
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
+                         Table::Create(fs, dir->Str(), Identifier("db1", "tbl1$branch_b1")));
+    ASSERT_EQ("tbl1$branch_b1", table->Name());
+    ASSERT_EQ((std::vector<std::string>{"b1_c0"}), table->LatestSchema()->FieldNames());
+    for (const auto& main_identifier :
+         {Identifier("db1", "tbl1$branch_main"), Identifier("db1", "tbl1$branch_MAIN")}) {
+        ASSERT_OK_AND_ASSIGN(table, Table::Create(fs, dir->Str(), main_identifier));
+        ASSERT_EQ((std::vector<std::string>{"main_c0"}), table->LatestSchema()->FieldNames());
+    }
+
+    Status missing_branch =
+        Table::Create(fs, dir->Str(), Identifier("db1", "tbl1$branch_missing")).status();
+    ASSERT_TRUE(missing_branch.IsNotExist()) << missing_branch.ToString();
+    ASSERT_NOK_WITH_MSG(Table::Create(fs, dir->Str(), Identifier("db1", "tbl1$branch_../outside")),
+                        "branch");
+}
+
 TEST(TableTest, TestCreateRejectsATableTypeThatIsNotAManagedTable) {
     // A `Table` describes a managed paimon table. Handing back another type as one would promise
     // snapshots and manifests it never had, so the failure belongs at the call that opens the