Skip to content

Commit 887e356

Browse files
suxiaogang223suxiaogang223
authored
feat: add system table framework with options table (#261)
1 parent ea4c852 commit 887e356

25 files changed

Lines changed: 1091 additions & 33 deletions

‎include/paimon/catalog/identifier.h‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,10 @@
1616

1717
#pragma once
1818

19+
#include <optional>
1920
#include <string>
2021

22+
#include "paimon/result.h"
2123
#include "paimon/type_fwd.h"
2224
#include "paimon/visibility.h"
2325

@@ -27,18 +29,32 @@ namespace paimon {
2729
class PAIMON_EXPORT Identifier {
2830
public:
2931
static const char kUnknownDatabase[];
32+
static const char kSystemTableSplitter[];
33+
static const char kSystemBranchPrefix[];
34+
static const char kDefaultMainBranch[];
3035

3136
explicit Identifier(const std::string& table);
3237
Identifier(const std::string& database, const std::string& table);
3338

3439
bool operator==(const Identifier& other);
3540
const std::string& GetDatabaseName() const;
3641
const std::string& GetTableName() const;
42+
Result<std::string> GetDataTableName() const;
43+
Result<std::optional<std::string>> GetBranchName() const;
44+
Result<std::string> GetBranchNameOrDefault() const;
45+
Result<std::optional<std::string>> GetSystemTableName() const;
46+
Result<bool> IsSystemTable() const;
3747
std::string ToString() const;
3848

3949
private:
50+
Status SplitTableName() const;
51+
4052
const std::string database_;
4153
const std::string table_;
54+
mutable bool parsed_ = false;
55+
mutable std::string data_table_;
56+
mutable std::optional<std::string> branch_;
57+
mutable std::optional<std::string> system_table_;
4258
};
4359

4460
} // namespace paimon

‎include/paimon/schema/schema.h‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,16 @@ class PAIMON_EXPORT Schema {
5757
/// failure.
5858
virtual Result<FieldType> GetFieldType(const std::string& field_name) const = 0;
5959

60+
/// Get an optional comment describing the schema object.
61+
/// @return The comment if set, or std::nullopt otherwise.
62+
virtual std::optional<std::string> Comment() const = 0;
63+
};
64+
65+
/// Schema contract for data tables.
66+
class PAIMON_EXPORT DataSchema : public Schema {
67+
public:
68+
~DataSchema() override = default;
69+
6070
/// Get the unique identifier of this table schema.
6171
/// @return The schema id
6272
virtual int64_t Id() const = 0;
@@ -86,10 +96,12 @@ class PAIMON_EXPORT Schema {
8696
/// Get the table-level options associated with this schema.
8797
/// @return A reference to the map of option key-value pairs (e.g., file format, filesystem).
8898
virtual const std::map<std::string, std::string>& Options() const = 0;
99+
};
89100

90-
/// Get an optional comment describing the table.
91-
/// @return The table comment if set, or std::nullopt otherwise.
92-
virtual std::optional<std::string> Comment() const = 0;
101+
/// Schema contract for system tables.
102+
class PAIMON_EXPORT SystemSchema : public Schema {
103+
public:
104+
~SystemSchema() override = default;
93105
};
94106

95107
} // namespace paimon

‎include/paimon/table/source/table_read.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,10 @@ class PAIMON_EXPORT TableRead {
6666
protected:
6767
explicit TableRead(const std::shared_ptr<MemoryPool>& memory_pool);
6868

69+
std::shared_ptr<MemoryPool> GetMemoryPool() const {
70+
return pool_;
71+
}
72+
6973
private:
7074
std::shared_ptr<MemoryPool> pool_;
7175
};

‎src/paimon/CMakeLists.txt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -314,6 +314,11 @@ set(PAIMON_CORE_SRCS
314314
core/table/source/table_read.cpp
315315
core/table/source/table_scan.cpp
316316
core/table/source/data_evolution_batch_scan.cpp
317+
core/table/system/options_system_table.cpp
318+
core/table/system/system_table.cpp
319+
core/table/system/system_table_read.cpp
320+
core/table/system/system_table_scan.cpp
321+
core/table/system/system_table_schema.cpp
317322
core/tag/tag.cpp
318323
core/utils/field_mapping.cpp
319324
core/utils/file_store_path_factory.cpp

‎src/paimon/core/catalog/file_system_catalog.cpp‎

Lines changed: 57 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
#include "paimon/common/utils/path_util.h"
3131
#include "paimon/common/utils/string_utils.h"
3232
#include "paimon/core/snapshot.h"
33+
#include "paimon/core/table/system/system_table.h"
34+
#include "paimon/core/table/system/system_table_schema.h"
3335
#include "paimon/core/utils/branch_manager.h"
3436
#include "paimon/core/utils/snapshot_manager.h"
3537
#include "paimon/defs.h"
@@ -92,6 +94,19 @@ Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const
9294
}
9395

9496
Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const {
97+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
98+
if (is_system_table) {
99+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name,
100+
identifier.GetSystemTableName());
101+
if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) {
102+
return false;
103+
}
104+
PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
105+
Identifier data_identifier(identifier.GetDatabaseName(), data_table_name);
106+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
107+
TableSchemaExists(data_identifier));
108+
return latest_schema != std::nullopt;
109+
}
95110
PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
96111
TableSchemaExists(identifier));
97112
return latest_schema != std::nullopt;
@@ -110,7 +125,8 @@ Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema*
110125
const std::vector<std::string>& primary_keys,
111126
const std::map<std::string, std::string>& options,
112127
bool ignore_if_exists) {
113-
if (IsSystemTable(identifier)) {
128+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier));
129+
if (is_system_table) {
114130
return Status::Invalid(
115131
fmt::format("Cannot create table for system table {}, please use data table.",
116132
identifier.ToString()));
@@ -148,7 +164,8 @@ Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema*
148164

149165
Result<std::optional<std::shared_ptr<TableSchema>>> FileSystemCatalog::TableSchemaExists(
150166
const Identifier& identifier) const {
151-
if (IsSystemTable(identifier)) {
167+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier));
168+
if (is_system_table) {
152169
return Status::NotImplemented(
153170
"do not support checking TableSchemaExists for system table.");
154171
}
@@ -168,12 +185,15 @@ bool FileSystemCatalog::IsSystemDatabase(const std::string& db_name) {
168185
return db_name == SYSTEM_DATABASE_NAME;
169186
}
170187

171-
bool FileSystemCatalog::IsSpecifiedSystemTable(const Identifier& identifier) {
172-
return (identifier.GetTableName().find(SYSTEM_TABLE_SPLITTER) != std::string::npos);
188+
Result<bool> FileSystemCatalog::IsSpecifiedSystemTable(const Identifier& identifier) {
189+
return identifier.IsSystemTable();
173190
}
174191

175-
bool FileSystemCatalog::IsSystemTable(const Identifier& identifier) {
176-
return IsSystemDatabase(identifier.GetDatabaseName()) || IsSpecifiedSystemTable(identifier);
192+
Result<bool> FileSystemCatalog::IsSystemTable(const Identifier& identifier) {
193+
if (IsSystemDatabase(identifier.GetDatabaseName())) {
194+
return true;
195+
}
196+
return IsSpecifiedSystemTable(identifier);
177197
}
178198

179199
std::string FileSystemCatalog::NewDatabasePath(const std::string& warehouse,
@@ -237,6 +257,26 @@ Result<bool> FileSystemCatalog::TableExistsInFileSystem(const std::string& table
237257

238258
Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
239259
const Identifier& identifier) const {
260+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
261+
if (is_system_table) {
262+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name,
263+
identifier.GetSystemTableName());
264+
if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) {
265+
return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
266+
}
267+
PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
268+
Identifier data_identifier(identifier.GetDatabaseName(), data_table_name);
269+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
270+
TableSchemaExists(data_identifier));
271+
if (!latest_schema) {
272+
return Status::NotExist(fmt::format("{} not exist", data_identifier.ToString()));
273+
}
274+
PAIMON_ASSIGN_OR_RAISE(
275+
std::shared_ptr<SystemTable> system_table,
276+
SystemTableLoader::Load(system_table_name.value(), fs_, GetTableLocation(identifier),
277+
latest_schema.value()));
278+
return std::make_shared<SystemTableSchema>(system_table->ArrowSchema());
279+
}
240280
PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
241281
TableSchemaExists(identifier));
242282
if (!latest_schema) {
@@ -246,6 +286,12 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
246286
}
247287

248288
Result<std::shared_ptr<Table>> FileSystemCatalog::GetTable(const Identifier& identifier) const {
289+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
290+
if (is_system_table) {
291+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, LoadTableSchema(identifier));
292+
return std::make_shared<Table>(schema, identifier.GetDatabaseName(),
293+
identifier.GetTableName());
294+
}
249295
return Table::Create(fs_, GetTableLocation(identifier), identifier);
250296
}
251297

@@ -351,7 +397,8 @@ Status FileSystemCatalog::DropTableImpl(const Identifier& identifier,
351397
}
352398

353399
Status FileSystemCatalog::DropTable(const Identifier& identifier, bool ignore_if_not_exists) {
354-
if (IsSystemTable(identifier)) {
400+
PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier));
401+
if (is_system_table) {
355402
return Status::Invalid(fmt::format("Cannot drop system table {}.", identifier.ToString()));
356403
}
357404

@@ -414,7 +461,9 @@ Status FileSystemCatalog::DropTable(const Identifier& identifier, bool ignore_if
414461

415462
Status FileSystemCatalog::RenameTable(const Identifier& from_table, const Identifier& to_table,
416463
bool ignore_if_not_exists) {
417-
if (IsSystemTable(from_table) || IsSystemTable(to_table)) {
464+
PAIMON_ASSIGN_OR_RAISE(bool is_from_system_table, IsSystemTable(from_table));
465+
PAIMON_ASSIGN_OR_RAISE(bool is_to_system_table, IsSystemTable(to_table));
466+
if (is_from_system_table || is_to_system_table) {
418467
return Status::Invalid(fmt::format("Cannot rename system table {} or {}.",
419468
from_table.ToString(), to_table.ToString()));
420469
}

‎src/paimon/core/catalog/file_system_catalog.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,8 +69,8 @@ class FileSystemCatalog : public Catalog {
6969
static std::string NewDatabasePath(const std::string& warehouse, const std::string& db_name);
7070
static std::string NewDataTablePath(const std::string& warehouse, const Identifier& identifier);
7171
static bool IsSystemDatabase(const std::string& db_name);
72-
static bool IsSpecifiedSystemTable(const Identifier& identifier);
73-
static bool IsSystemTable(const Identifier& identifier);
72+
static Result<bool> IsSpecifiedSystemTable(const Identifier& identifier);
73+
static Result<bool> IsSystemTable(const Identifier& identifier);
7474
Result<std::optional<std::shared_ptr<TableSchema>>> TableSchemaExists(
7575
const Identifier& identifier) const;
7676

‎src/paimon/core/catalog/file_system_catalog_test.cpp‎

Lines changed: 71 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
#include "paimon/common/data/blob_utils.h"
2626
#include "paimon/common/utils/path_util.h"
2727
#include "paimon/core/core_options.h"
28+
#include "paimon/core/schema/table_schema.h"
29+
#include "paimon/core/table/system/system_table_schema.h"
2830
#include "paimon/defs.h"
2931
#include "paimon/fs/file_system.h"
3032
#include "paimon/fs/file_system_factory.h"
@@ -149,6 +151,65 @@ TEST(FileSystemCatalogTest, TestCreateTable) {
149151
ArrowSchemaRelease(&schema);
150152
}
151153

154+
TEST(FileSystemCatalogTest, TestOptionsSystemTableCatalog) {
155+
std::map<std::string, std::string> options;
156+
options[Options::FILE_SYSTEM] = "local";
157+
options[Options::FILE_FORMAT] = "orc";
158+
options["custom.option"] = "custom-value";
159+
ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
160+
auto dir = UniqueTestDirectory::Create();
161+
ASSERT_TRUE(dir);
162+
FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str());
163+
ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true));
164+
165+
auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())});
166+
::ArrowSchema schema;
167+
ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok());
168+
ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema,
169+
/*partition_keys=*/{}, /*primary_keys=*/{}, options,
170+
/*ignore_if_exists=*/false));
171+
ArrowSchemaRelease(&schema);
172+
173+
Identifier options_identifier("db1", "tbl1$options");
174+
ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(options_identifier));
175+
ASSERT_TRUE(exists);
176+
ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "tbl1$unknown")));
177+
ASSERT_FALSE(exists);
178+
ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "missing$options")));
179+
ASSERT_FALSE(exists);
180+
ASSERT_EQ(catalog.GetTableLocation(options_identifier),
181+
PathUtil::JoinPath(PathUtil::JoinPath(dir->Str(), "db1.db"), "tbl1$options"));
182+
183+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> system_schema,
184+
catalog.LoadTableSchema(options_identifier));
185+
ASSERT_TRUE(std::dynamic_pointer_cast<SystemSchema>(system_schema) != nullptr);
186+
ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(system_schema) != nullptr);
187+
ASSERT_OK_AND_ASSIGN(auto c_schema, system_schema->GetArrowSchema());
188+
auto loaded_schema_result = arrow::ImportSchema(c_schema.get());
189+
ASSERT_TRUE(loaded_schema_result.ok()) << loaded_schema_result.status().ToString();
190+
auto loaded_schema = loaded_schema_result.ValueUnsafe();
191+
ASSERT_EQ(loaded_schema->field_names(), (std::vector<std::string>{"key", "value"}));
192+
ASSERT_EQ(loaded_schema->field(0)->type()->id(), arrow::Type::STRING);
193+
ASSERT_EQ(loaded_schema->field(1)->type()->id(), arrow::Type::STRING);
194+
ASSERT_FALSE(loaded_schema->field(0)->nullable());
195+
ASSERT_FALSE(loaded_schema->field(1)->nullable());
196+
197+
ASSERT_OK_AND_ASSIGN(auto system_table, catalog.GetTable(options_identifier));
198+
ASSERT_EQ(system_table->Name(), "tbl1$options");
199+
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl1$unknown")), "not exist");
200+
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "missing$options")), "not exist");
201+
202+
::ArrowSchema system_create_schema;
203+
ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok());
204+
ASSERT_NOK_WITH_MSG(
205+
catalog.CreateTable(options_identifier, &system_create_schema, {}, {}, options, false),
206+
"Cannot create table for system table");
207+
ArrowSchemaRelease(&system_create_schema);
208+
ASSERT_NOK_WITH_MSG(catalog.DropTable(options_identifier, false), "Cannot drop system table");
209+
ASSERT_NOK_WITH_MSG(catalog.RenameTable(options_identifier, Identifier("db1", "tbl2"), false),
210+
"Cannot rename system table");
211+
}
212+
152213
TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
153214
std::map<std::string, std::string> options;
154215
options[Options::FILE_SYSTEM] = "local";
@@ -188,6 +249,8 @@ TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
188249
ASSERT_EQ(table_names[0], "tbl1");
189250
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> table_schema,
190251
catalog.LoadTableSchema(Identifier("db1", "tbl1")));
252+
ASSERT_TRUE(std::dynamic_pointer_cast<DataSchema>(table_schema) != nullptr);
253+
ASSERT_TRUE(std::dynamic_pointer_cast<TableSchema>(table_schema) != nullptr);
191254
ASSERT_OK_AND_ASSIGN(auto arrow_schema, table_schema->GetArrowSchema());
192255
auto loaded_schema = arrow::ImportSchema(arrow_schema.get()).ValueOrDie();
193256
ASSERT_TRUE(typed_schema.Equals(loaded_schema));
@@ -358,11 +421,13 @@ TEST(FileSystemCatalogTest, TestValidateTableSchema) {
358421
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db0", "tbl0")),
359422
"Identifier{database=\'db0\', table=\'tbl0\'} not exist");
360423
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> table_schema, catalog.LoadTableSchema(identifier));
361-
ASSERT_EQ(0, table_schema->Id());
362-
ASSERT_EQ(3, table_schema->HighestFieldId());
363-
ASSERT_EQ(1, table_schema->PartitionKeys().size());
364-
ASSERT_EQ(0, table_schema->PrimaryKeys().size());
365-
ASSERT_EQ(-1, table_schema->NumBuckets());
424+
auto data_schema = std::dynamic_pointer_cast<DataSchema>(table_schema);
425+
ASSERT_TRUE(data_schema != nullptr);
426+
ASSERT_EQ(0, data_schema->Id());
427+
ASSERT_EQ(3, data_schema->HighestFieldId());
428+
ASSERT_EQ(1, data_schema->PartitionKeys().size());
429+
ASSERT_EQ(0, data_schema->PrimaryKeys().size());
430+
ASSERT_EQ(-1, data_schema->NumBuckets());
366431
ASSERT_FALSE(table_schema->Comment().has_value());
367432
std::vector<std::string> field_names = table_schema->FieldNames();
368433
std::vector<std::string> expected_field_names = {"f0", "f1", "f2", "f3"};
@@ -396,8 +461,7 @@ TEST(FileSystemCatalogTest, TestValidateTableSchema) {
396461
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(identifier),
397462
"Identifier{database=\'db1\', table=\'tbl1\'} not exist");
398463

399-
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl$11")),
400-
"do not support checking TableSchemaExists for system table.");
464+
ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl$11")), "not exist");
401465
ArrowSchemaRelease(&schema);
402466
}
403467

0 commit comments

Comments
 (0)