Skip to content

Commit dddfda5

Browse files
authored
feat(table): make Table a resolved handle and the home of operator creation (#434)
1 parent fff5a58 commit dddfda5

31 files changed

Lines changed: 1965 additions & 1682 deletions

‎examples/read_write_demo.cpp‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#include "arrow/ipc/api.h"
2727
#include "paimon/api.h"
2828
#include "paimon/catalog/catalog.h"
29+
#include "paimon/catalog/table.h"
2930

3031
arrow::Result<std::shared_ptr<arrow::StructArray>> PrepareData(const arrow::FieldVector& fields) {
3132
arrow::StringBuilder f0_builder;
@@ -81,6 +82,11 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
8182
/*primary_keys=*/{}, options,
8283
/*ignore_if_exists=*/false));
8384

85+
// The table every operator below is created from. Each call hands the table's shared owner
86+
// to the operator it creates.
87+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Table> table,
88+
catalog->GetTable(paimon::Identifier(db_name, table_name)));
89+
8490
std::string table_path = root_path + "/" + db_name + ".db/" + table_name;
8591

8692
std::string commit_user = "some_commit_user";
@@ -89,7 +95,7 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
8995
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::WriteContext> write_context,
9096
context_builder.SetOptions(options).Finish());
9197
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreWrite> writer,
92-
paimon::FileStoreWrite::Create(std::move(write_context)));
98+
table->NewWrite(std::move(write_context)));
9399

94100
// prepare data
95101
auto struct_array = PrepareData(fields);
@@ -113,23 +119,23 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
113119
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::CommitContext> commit_context,
114120
commit_context_builder.SetOptions(options).Finish());
115121
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreCommit> committer,
116-
paimon::FileStoreCommit::Create(std::move(commit_context)));
122+
table->NewCommit(std::move(commit_context)));
117123
PAIMON_RETURN_NOT_OK(committer->Commit(commit_message));
118124

119125
// scan
120126
paimon::ScanContextBuilder scan_context_builder(table_path);
121127
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ScanContext> scan_context,
122128
scan_context_builder.SetOptions(options).Finish());
123129
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableScan> scanner,
124-
paimon::TableScan::Create(std::move(scan_context)));
130+
table->NewScan(std::move(scan_context)));
125131
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Plan> plan, scanner->CreatePlan());
126132

127133
// read
128134
paimon::ReadContextBuilder read_context_builder(table_path);
129135
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ReadContext> read_context,
130136
read_context_builder.SetOptions(options).Finish());
131137
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableRead> table_read,
132-
paimon::TableRead::Create(std::move(read_context)));
138+
table->NewRead(std::move(read_context)));
133139
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::BatchReader> batch_reader,
134140
table_read->CreateReader(plan->Splits()));
135141
arrow::ArrayVector result_array_vector;

‎include/paimon/catalog/identifier.h‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,16 +27,16 @@
2727

2828
namespace paimon {
2929

30-
/// An identifier for a table containing database and table name.
30+
/// An identifier for a catalog object: a database name plus an object name, which may carry a
31+
/// branch or system-table suffix.
3132
class PAIMON_EXPORT Identifier {
3233
public:
3334
static const char kUnknownDatabase[];
3435
static const char kSystemTableSplitter[];
3536
static const char kSystemBranchPrefix[];
3637
static const char kDefaultMainBranch[];
3738

38-
explicit Identifier(const std::string& table);
39-
Identifier(const std::string& database, const std::string& table);
39+
Identifier(const std::string& database, const std::string& object);
4040
/// Names `branch` of `table`: every branch but the main one is addressed by the object name
4141
/// this builds, which joins them as `tbl$branch_dev`.
4242
///
@@ -53,9 +53,15 @@ class PAIMON_EXPORT Identifier {
5353

5454
bool operator==(const Identifier& other) const;
5555
const std::string& GetDatabaseName() const;
56-
const std::string& GetTableName() const;
56+
/// The full object name, which keeps any branch and system-table suffix, like
57+
/// `tbl$branch_dev`. Mirrors the Java `Identifier#getObjectName`.
58+
const std::string& GetObjectName() const;
5759
Result<std::string> GetDataTableName() const;
5860
Result<std::optional<std::string>> GetBranchName() const;
61+
/// The branch the object name carries, `main` when it names none. Any spelling of the
62+
/// main branch, as `MAIN`, folds to `main` here, as the branch constructor folds it,
63+
/// so a caller binding a manager to the returned value addresses the main branch's own
64+
/// directory rather than one named after the alias.
5965
Result<std::string> GetBranchNameOrDefault() const;
6066
Result<std::optional<std::string>> GetSystemTableName() const;
6167
Result<bool> IsSystemTable() const;
@@ -67,10 +73,10 @@ class PAIMON_EXPORT Identifier {
6773
static Result<Identifier> FromString(const std::string& full_name);
6874

6975
private:
70-
Status SplitTableName() const;
76+
Status SplitObjectName() const;
7177

7278
const std::string database_;
73-
const std::string table_;
79+
const std::string object_;
7480
mutable bool parsed_ = false;
7581
mutable std::string data_table_;
7682
mutable std::optional<std::string> branch_;

‎include/paimon/catalog/table.h‎

Lines changed: 61 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -33,33 +33,50 @@
3333

3434
namespace paimon {
3535

36+
class TableScan;
37+
class TableRead;
38+
3639
/// A table provides basic abstraction for table type.
3740
class PAIMON_EXPORT Table {
3841
public:
39-
/// Creates a table by loading the latest schema from the specified table path.
40-
///
41-
/// @param file_system File system used to access the table directory and schema files.
42-
/// @param table_path Root path of the table in the file system.
43-
/// @param identifier Logical table identifier selecting the schema branch and table name.
44-
/// @return A table initialized with the latest available schema.
45-
static Result<std::shared_ptr<Table>> Create(const std::shared_ptr<FileSystem>& file_system,
46-
const std::string& table_path,
47-
const Identifier& identifier);
48-
42+
/// @param file_system File system the table's data is accessed through.
43+
/// @param root_path Root path of the table in the file system.
4944
/// @param uuid Metastore UUID, or empty if unavailable.
50-
Table(const std::shared_ptr<Schema>& schema, const std::string& database,
51-
const std::string& table_name, const std::string& uuid = "")
52-
: schema_(schema), database_(database), table_name_(table_name), uuid_(uuid) {}
45+
Table(const std::shared_ptr<FileSystem>& file_system, const std::string& root_path,
46+
const std::shared_ptr<Schema>& schema, const Identifier& identifier,
47+
const std::string& uuid = "");
5348

5449
~Table() = default;
5550

51+
/// Returns the file system the table was opened with, which serves the table's
52+
/// files. A catalog that hands out per-table file systems opens a table with the
53+
/// one for that table; it is null when the opener had none.
54+
std::shared_ptr<FileSystem> GetFileSystem() const {
55+
return file_system_;
56+
}
57+
58+
/// Returns the root path of the table in its file system. It is empty when the
59+
/// opener had no location for the table, which a mock in a test is one case of.
60+
const std::string& GetRootPath() const {
61+
return root_path_;
62+
}
63+
5664
/// A name to identify this table.
5765
std::string Name() const {
58-
return table_name_;
66+
return identifier_.GetObjectName();
5967
}
6068

6169
/// Full name of the table, default is database.tableName.
62-
std::string FullName() const;
70+
std::string FullName() const {
71+
return identifier_.GetFullName();
72+
}
73+
74+
/// Returns the branch the table was opened for, which its identifier carries; `main`
75+
/// when it names none or a spelling of it, which its managers are bound to. A
76+
/// malformed object name that hides the branch is reported.
77+
Result<std::string> CurrentBranch() const {
78+
return identifier_.GetBranchNameOrDefault();
79+
}
6380

6481
/// Returns the metastore UUID, falling back to `FullName()`.
6582
/// @note Use `CatalogUuid()` for commit requests to avoid sending a table name as an ID.
@@ -78,10 +95,37 @@ class PAIMON_EXPORT Table {
7895
return schema_;
7996
}
8097

98+
/// Creates a scan of this table from a finished scan context, the table-first entry point
99+
/// that `TableScan::Create()` also builds through. The scan is derived from the context.
100+
/// @param context The finished scan context.
101+
/// @return The scan, or the error building it reports.
102+
Result<std::unique_ptr<TableScan>> NewScan(std::unique_ptr<ScanContext> context) const;
103+
104+
/// Creates a read of this table from a finished read context, the table-first entry point
105+
/// that `TableRead::Create()` also builds through. The read is derived from the context.
106+
/// @param context The finished read context.
107+
/// @return The read, or the error building it reports.
108+
Result<std::unique_ptr<TableRead>> NewRead(std::unique_ptr<ReadContext> context) const;
109+
110+
/// Creates a write of this table from a finished write context, the table-first entry point
111+
/// that `FileStoreWrite::Create()` also builds through. The write is derived from the context.
112+
/// @param context The finished write context.
113+
/// @return The write, or the error building it reports.
114+
Result<std::unique_ptr<FileStoreWrite>> NewWrite(std::unique_ptr<WriteContext> context) const;
115+
116+
/// Creates a commit of this table from a finished commit context, the table-first entry point
117+
/// that `FileStoreCommit::Create()` also builds through. The commit is derived from the
118+
/// context.
119+
/// @param context The finished commit context.
120+
/// @return The commit, or the error building it reports.
121+
Result<std::unique_ptr<FileStoreCommit>> NewCommit(
122+
std::unique_ptr<CommitContext> context) const;
123+
81124
private:
125+
std::shared_ptr<FileSystem> file_system_;
126+
std::string root_path_;
82127
std::shared_ptr<Schema> schema_;
83-
std::string database_;
84-
std::string table_name_;
128+
Identifier identifier_;
85129
std::string uuid_;
86130
};
87131

‎include/paimon/read_context.h‎

Lines changed: 29 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -52,24 +52,23 @@ struct VectorSearch;
5252
/// @see ReadContextBuilder
5353
class PAIMON_EXPORT ReadContext {
5454
public:
55-
ReadContext(const std::string& path, const std::string& branch,
56-
const std::vector<std::string>& read_field_names,
57-
const std::vector<int32_t>& read_field_ids,
58-
const std::shared_ptr<Predicate>& predicate,
59-
const std::shared_ptr<VectorSearch>& vector_search,
60-
const std::shared_ptr<FullTextSearch>& full_text_search,
61-
bool enable_predicate_filter, bool enable_prefetch, bool enable_late_materializing,
62-
uint32_t prefetch_batch_count, uint32_t prefetch_max_parallel_num,
63-
bool enable_multi_thread_row_to_batch, uint32_t row_to_batch_thread_number,
64-
const std::optional<std::string>& table_schema,
65-
const std::shared_ptr<MemoryPool>& memory_pool,
66-
const std::shared_ptr<Executor>& executor,
67-
const std::shared_ptr<FileSystem>& specific_file_system,
68-
const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
69-
const std::shared_ptr<RealtimeContext>& realtime_context,
70-
const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
71-
const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
72-
const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level);
55+
ReadContext(
56+
const std::string& path, const std::string& branch,
57+
const std::vector<std::string>& read_field_names,
58+
const std::vector<int32_t>& read_field_ids, const std::shared_ptr<Predicate>& predicate,
59+
const std::shared_ptr<VectorSearch>& vector_search,
60+
const std::shared_ptr<FullTextSearch>& full_text_search, bool enable_predicate_filter,
61+
bool enable_prefetch, bool enable_late_materializing, uint32_t prefetch_batch_count,
62+
uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
63+
uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
64+
const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor,
65+
const std::shared_ptr<FileSystem>& specific_file_system,
66+
const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
67+
const std::shared_ptr<RealtimeContext>& realtime_context,
68+
const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
69+
const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
70+
const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level,
71+
const std::shared_ptr<Catalog>& catalog, const std::optional<Identifier>& identifier);
7372

7473
~ReadContext();
7574

@@ -174,6 +173,16 @@ class PAIMON_EXPORT ReadContext {
174173
return warmup_level_;
175174
}
176175

176+
/// Returns the catalog supplying table metadata, or null if unset.
177+
const std::shared_ptr<Catalog>& GetCatalog() const {
178+
return catalog_;
179+
}
180+
181+
/// Returns the table identifier supplied with the catalog.
182+
const std::optional<Identifier>& GetIdentifier() const {
183+
return identifier_;
184+
}
185+
177186
/// Whether a read schema (C ArrowSchema) for nested column pruning was provided.
178187
bool HasReadSchema() const {
179188
return read_schema_ != nullptr && read_schema_->release != nullptr;
@@ -217,6 +226,8 @@ class PAIMON_EXPORT ReadContext {
217226
std::shared_ptr<Cache> cache_;
218227
std::shared_ptr<FormatTable> format_table_;
219228
WarmupLevel warmup_level_;
229+
std::shared_ptr<Catalog> catalog_;
230+
std::optional<Identifier> identifier_;
220231
// Owns schema resources and releases ArrowSchema::release in destructor.
221232
std::unique_ptr<ArrowSchema> read_schema_;
222233
};

‎include/paimon/scan_context.h‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,9 @@ class PAIMON_EXPORT ScanContext {
5858
const std::optional<std::string>& table_schema,
5959
const std::map<std::string, std::string>& options,
6060
const std::shared_ptr<Cache>& cache,
61-
const std::shared_ptr<FormatTable>& format_table);
61+
const std::shared_ptr<FormatTable>& format_table,
62+
const std::shared_ptr<Catalog>& catalog,
63+
const std::optional<Identifier>& identifier);
6264

6365
~ScanContext();
6466

@@ -115,6 +117,16 @@ class PAIMON_EXPORT ScanContext {
115117
return format_table_;
116118
}
117119

120+
/// Returns the catalog supplying table metadata, or null if unset.
121+
const std::shared_ptr<Catalog>& GetCatalog() const {
122+
return catalog_;
123+
}
124+
125+
/// Returns the table identifier supplied with the catalog.
126+
const std::optional<Identifier>& GetIdentifier() const {
127+
return identifier_;
128+
}
129+
118130
private:
119131
std::string path_;
120132
bool is_streaming_mode_;
@@ -129,6 +141,8 @@ class PAIMON_EXPORT ScanContext {
129141
std::map<std::string, std::string> options_;
130142
std::shared_ptr<Cache> cache_;
131143
std::shared_ptr<FormatTable> format_table_;
144+
std::shared_ptr<Catalog> catalog_;
145+
std::optional<Identifier> identifier_;
132146
};
133147

134148
/// Filter configuration for table scan operations

‎include/paimon/table/format/format_table.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ class PAIMON_EXPORT FormatTable {
157157

158158
/// A name to identify this table.
159159
std::string Name() const {
160-
return identifier_.GetTableName();
160+
return identifier_.GetObjectName();
161161
}
162162

163163
/// Full name of the table, database.tableName.

‎src/paimon/common/data/blob_view_struct_test.cpp‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ class BlobViewStructTest : public testing::Test {
4040

4141
TEST_F(BlobViewStructTest, TestConstructorAndGetters) {
4242
ASSERT_EQ(view_struct_.GetIdentifier().GetDatabaseName(), "test_db");
43-
ASSERT_EQ(view_struct_.GetIdentifier().GetTableName(), "test_table");
43+
ASSERT_EQ(view_struct_.GetIdentifier().GetObjectName(), "test_table");
4444
ASSERT_EQ(view_struct_.FieldId(), 7);
4545
ASSERT_EQ(view_struct_.RowId(), 1024);
4646
}
@@ -53,7 +53,7 @@ TEST_F(BlobViewStructTest, TestSerializeDeserializeRoundTrip) {
5353
ASSERT_OK_AND_ASSIGN(auto restored,
5454
BlobViewStruct::Deserialize(serialized->data(), serialized->size()));
5555
ASSERT_EQ(restored->GetIdentifier().GetDatabaseName(), "test_db");
56-
ASSERT_EQ(restored->GetIdentifier().GetTableName(), "test_table");
56+
ASSERT_EQ(restored->GetIdentifier().GetObjectName(), "test_table");
5757
ASSERT_EQ(restored->FieldId(), 7);
5858
ASSERT_EQ(restored->RowId(), 1024);
5959
}
@@ -78,7 +78,8 @@ TEST_F(BlobViewStructTest, TestToString) {
7878
}
7979

8080
TEST_F(BlobViewStructTest, TestSerializeRejectsUnknownDatabase) {
81-
BlobViewStruct view_struct(Identifier("test_table"), /*field_id=*/7, /*row_id=*/1024);
81+
BlobViewStruct view_struct(Identifier(Identifier::kUnknownDatabase, "test_table"),
82+
/*field_id=*/7, /*row_id=*/1024);
8283
ASSERT_NOK_WITH_MSG(view_struct.Serialize(pool_),
8384
"upstream table identifier must include database name");
8485
}

‎src/paimon/common/data/blob_view_test.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,8 @@ TEST_F(BlobViewTest, TestJavaSerializationCompatibility) {
8080
}
8181

8282
TEST_F(BlobViewTest, TestToViewRejectsInvalidArguments) {
83-
BlobView unknown_database{Identifier("source"), /*field_id=*/7, /*row_id=*/5};
83+
BlobView unknown_database{Identifier(Identifier::kUnknownDatabase, "source"),
84+
/*field_id=*/7, /*row_id=*/5};
8485
ASSERT_NOK_WITH_MSG(BlobView::ToView(unknown_database, pool_),
8586
"upstream table identifier must include database name");
8687
ASSERT_NOK_WITH_MSG(BlobView::ToView(blob_view_, /*pool=*/nullptr), "memory pool is nullptr");

0 commit comments

Comments
 (0)