From 948ed6245111d7d1f9a716d4ee600e6895cd2057 Mon Sep 17 00:00:00 2001 From: mrdrivingduck Date: Mon, 12 Oct 2026 17:31:21 +0800 Subject: [PATCH] fix(catalog): load schemas from the requested branch Preserve branch selection when loading table metadata so normal and system tables use the schema of the requested branch. Apply the same selection when opening tables directly, including default-branch aliases. Reject missing branches instead of falling back to the main schema. --- include/paimon/catalog/table.h | 2 +- .../core/catalog/file_system_catalog.cpp | 22 ++++-- .../core/catalog/file_system_catalog_test.cpp | 73 +++++++++++++++++++ src/paimon/core/table/table.cpp | 8 +- src/paimon/core/table/table_test.cpp | 30 ++++++++ 5 files changed, 125 insertions(+), 10 deletions(-) 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> Create(const std::shared_ptr& 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 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 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> latest_schema, TableSchemaExists(data_identifier)); return latest_schema != std::nullopt; @@ -181,7 +182,11 @@ Result>> 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> FileSystemCatalog::LoadTableSchema( system_table->ArrowSchema()); return std::make_shared(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> 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 branch, identifier.GetBranchName()); + PAIMON_ASSIGN_OR_RAISE( + Identifier data_identifier, + CatalogUtils::BranchIdentifier(identifier, branch.value_or(std::string()))); PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, TableSchemaExists(data_identifier)); if (!latest_schema) { return Status::NotExist(fmt::format("{} not exist", data_identifier.ToString())); } std::map dynamic_options; - PAIMON_ASSIGN_OR_RAISE(std::optional 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 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 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 loaded_schema, + catalog.LoadTableSchema(branch_identifier)); + auto data_schema = std::dynamic_pointer_cast(loaded_schema); + ASSERT_NE(nullptr, data_schema); + ASSERT_EQ((std::vector{"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, catalog.GetTable(branch_identifier)); + data_schema = std::dynamic_pointer_cast(table->LatestSchema()); + ASSERT_NE(nullptr, data_schema); + ASSERT_EQ((std::vector{"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(loaded_schema); + ASSERT_NE(nullptr, data_schema); + ASSERT_EQ((std::vector{"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{"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 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> Table::Create(const std::shared_ptr& 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> 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 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 branch_schema, + branch_schema_manager.CreateTable( + arrow::schema({arrow::field("b1_c0", arrow::int32())}), {}, {}, {})); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr
table, + Table::Create(fs, dir->Str(), Identifier("db1", "tbl1$branch_b1"))); + ASSERT_EQ("tbl1$branch_b1", table->Name()); + ASSERT_EQ((std::vector{"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{"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