This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new ff4194ec fix(catalog): load schemas from the requested branch (#429)
ff4194ec is described below
commit ff4194ec81c1b435f964fe757aa16a70d6043e76
Author: Mr Dk. <[email protected]>
AuthorDate: Thu Oct 15 08:51:23 2026 +0800
fix(catalog): load schemas from the requested branch (#429)
---
include/paimon/catalog/table.h | 2 +-
src/paimon/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 ef68d1e4..59c82495 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 09be18d7..827e50b4 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 74578e39..68aa6ac4 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 eaa57440..c627b458 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 60113ec8..8032890e 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
