feat(table): make Table a resolved handle and the home of operator creation

examples/read_write_demo.cpp

@@ -26,6 +26,7 @@
 #include "arrow/ipc/api.h"
 #include "paimon/api.h"
 #include "paimon/catalog/catalog.h"
+#include "paimon/catalog/table.h"
 
 arrow::Result<std::shared_ptr<arrow::StructArray>> PrepareData(const arrow::FieldVector& fields) {
     arrow::StringBuilder f0_builder;
@@ -81,6 +82,11 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
                                               /*primary_keys=*/{}, options,
                                               /*ignore_if_exists=*/false));
 
+    // The table every operator below is created from. Each call hands the table's shared owner
+    // to the operator it creates.
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Table> table,
+                           catalog->GetTable(paimon::Identifier(db_name, table_name)));
+
     std::string table_path = root_path + "/" + db_name + ".db/" + table_name;
 
     std::string commit_user = "some_commit_user";
@@ -89,7 +95,7 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::WriteContext> write_context,
                            context_builder.SetOptions(options).Finish());
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreWrite> writer,
-                           paimon::FileStoreWrite::Create(std::move(write_context)));
+                           table->NewWrite(std::move(write_context)));
 
     // prepare data
     auto struct_array = PrepareData(fields);
@@ -113,23 +119,23 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::CommitContext> commit_context,
                            commit_context_builder.SetOptions(options).Finish());
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreCommit> committer,
-                           paimon::FileStoreCommit::Create(std::move(commit_context)));
+                           table->NewCommit(std::move(commit_context)));
     PAIMON_RETURN_NOT_OK(committer->Commit(commit_message));
 
     // scan
     paimon::ScanContextBuilder scan_context_builder(table_path);
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ScanContext> scan_context,
                            scan_context_builder.SetOptions(options).Finish());
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableScan> scanner,
-                           paimon::TableScan::Create(std::move(scan_context)));
+                           table->NewScan(std::move(scan_context)));
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Plan> plan, scanner->CreatePlan());
 
     // read
     paimon::ReadContextBuilder read_context_builder(table_path);
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ReadContext> read_context,
                            read_context_builder.SetOptions(options).Finish());
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableRead> table_read,
-                           paimon::TableRead::Create(std::move(read_context)));
+                           table->NewRead(std::move(read_context)));
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::BatchReader> batch_reader,
                            table_read->CreateReader(plan->Splits()));
     arrow::ArrayVector result_array_vector;

include/paimon/catalog/identifier.h

@@ -27,16 +27,16 @@
 
 namespace paimon {
 
-/// An identifier for a table containing database and table name.
+/// An identifier for a catalog object: a database name plus an object name, which may carry a
+/// branch or system-table suffix.
 class PAIMON_EXPORT Identifier {
  public:
     static const char kUnknownDatabase[];
     static const char kSystemTableSplitter[];
     static const char kSystemBranchPrefix[];
     static const char kDefaultMainBranch[];
 
-    explicit Identifier(const std::string& table);
-    Identifier(const std::string& database, const std::string& table);
+    Identifier(const std::string& database, const std::string& object);
     /// Names `branch` of `table`: every branch but the main one is addressed by the object name
     /// this builds, which joins them as `tbl$branch_dev`.
     ///
@@ -53,9 +53,15 @@ class PAIMON_EXPORT Identifier {
 
     bool operator==(const Identifier& other) const;
     const std::string& GetDatabaseName() const;
-    const std::string& GetTableName() const;
+    /// The full object name, which keeps any branch and system-table suffix, like
+    /// `tbl$branch_dev`. Mirrors the Java `Identifier#getObjectName`.
+    const std::string& GetObjectName() const;
     Result<std::string> GetDataTableName() const;
     Result<std::optional<std::string>> GetBranchName() const;
+    /// The branch the object name carries, `main` when it names none. Any spelling of the
+    /// main branch, as `MAIN`, folds to `main` here, as the branch constructor folds it,
+    /// so a caller binding a manager to the returned value addresses the main branch's own
+    /// directory rather than one named after the alias.
     Result<std::string> GetBranchNameOrDefault() const;
     Result<std::optional<std::string>> GetSystemTableName() const;
     Result<bool> IsSystemTable() const;
@@ -67,10 +73,10 @@ class PAIMON_EXPORT Identifier {
     static Result<Identifier> FromString(const std::string& full_name);
 
  private:
-    Status SplitTableName() const;
+    Status SplitObjectName() const;
 
     const std::string database_;
-    const std::string table_;
+    const std::string object_;
     mutable bool parsed_ = false;
     mutable std::string data_table_;
     mutable std::optional<std::string> branch_;

include/paimon/catalog/table.h

@@ -33,33 +33,50 @@
 
 namespace paimon {
 
+class TableScan;
+class TableRead;
+
 /// A table provides basic abstraction for table type.
 class PAIMON_EXPORT Table {
  public:
-    /// Creates a table by loading the latest schema from the specified table path.
-    ///
-    /// @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 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,
-                                                 const Identifier& identifier);
-
+    /// @param file_system File system the table's data is accessed through.
+    /// @param root_path Root path of the table in the file system.
     /// @param uuid Metastore UUID, or empty if unavailable.
-    Table(const std::shared_ptr<Schema>& schema, const std::string& database,
-          const std::string& table_name, const std::string& uuid = "")
-        : schema_(schema), database_(database), table_name_(table_name), uuid_(uuid) {}
+    Table(const std::shared_ptr<FileSystem>& file_system, const std::string& root_path,
+          const std::shared_ptr<Schema>& schema, const Identifier& identifier,
+          const std::string& uuid = "");
 
     ~Table() = default;
 
+    /// Returns the file system the table was opened with, which serves the table's
+    /// files. A catalog that hands out per-table file systems opens a table with the
+    /// one for that table; it is null when the opener had none.
+    std::shared_ptr<FileSystem> GetFileSystem() const {
+        return file_system_;
+    }
+
+    /// Returns the root path of the table in its file system. It is empty when the
+    /// opener had no location for the table, which a mock in a test is one case of.
+    const std::string& GetRootPath() const {
+        return root_path_;
+    }
+
     /// A name to identify this table.
     std::string Name() const {
-        return table_name_;
+        return identifier_.GetObjectName();
     }
 
     /// Full name of the table, default is database.tableName.
-    std::string FullName() const;
+    std::string FullName() const {
+        return identifier_.GetFullName();
+    }
+
+    /// Returns the branch the table was opened for, which its identifier carries; `main`
+    /// when it names none or a spelling of it, which its managers are bound to. A
+    /// malformed object name that hides the branch is reported.
+    Result<std::string> CurrentBranch() const {
+        return identifier_.GetBranchNameOrDefault();
+    }
 
     /// Returns the metastore UUID, falling back to `FullName()`.
     /// @note Use `CatalogUuid()` for commit requests to avoid sending a table name as an ID.
@@ -78,10 +95,37 @@ class PAIMON_EXPORT Table {
         return schema_;
     }
 
+    /// Creates a scan of this table from a finished scan context, the table-first entry point
+    /// that `TableScan::Create()` also builds through. The scan is derived from the context.
+    /// @param context The finished scan context.
+    /// @return The scan, or the error building it reports.
+    Result<std::unique_ptr<TableScan>> NewScan(std::unique_ptr<ScanContext> context) const;
+
+    /// Creates a read of this table from a finished read context, the table-first entry point
+    /// that `TableRead::Create()` also builds through. The read is derived from the context.
+    /// @param context The finished read context.
+    /// @return The read, or the error building it reports.
+    Result<std::unique_ptr<TableRead>> NewRead(std::unique_ptr<ReadContext> context) const;
+
+    /// Creates a write of this table from a finished write context, the table-first entry point
+    /// that `FileStoreWrite::Create()` also builds through. The write is derived from the context.
+    /// @param context The finished write context.
+    /// @return The write, or the error building it reports.
+    Result<std::unique_ptr<FileStoreWrite>> NewWrite(std::unique_ptr<WriteContext> context) const;
+
+    /// Creates a commit of this table from a finished commit context, the table-first entry point
+    /// that `FileStoreCommit::Create()` also builds through. The commit is derived from the
+    /// context.
+    /// @param context The finished commit context.
+    /// @return The commit, or the error building it reports.
+    Result<std::unique_ptr<FileStoreCommit>> NewCommit(
+        std::unique_ptr<CommitContext> context) const;
+
  private:
+    std::shared_ptr<FileSystem> file_system_;
+    std::string root_path_;
     std::shared_ptr<Schema> schema_;
-    std::string database_;
-    std::string table_name_;
+    Identifier identifier_;
     std::string uuid_;
 };
 

include/paimon/read_context.h

@@ -52,24 +52,23 @@ struct VectorSearch;
 /// @see ReadContextBuilder
 class PAIMON_EXPORT ReadContext {
  public:
-    ReadContext(const std::string& path, const std::string& branch,
-                const std::vector<std::string>& read_field_names,
-                const std::vector<int32_t>& read_field_ids,
-                const std::shared_ptr<Predicate>& predicate,
-                const std::shared_ptr<VectorSearch>& vector_search,
-                const std::shared_ptr<FullTextSearch>& full_text_search,
-                bool enable_predicate_filter, bool enable_prefetch, bool enable_late_materializing,
-                uint32_t prefetch_batch_count, uint32_t prefetch_max_parallel_num,
-                bool enable_multi_thread_row_to_batch, uint32_t row_to_batch_thread_number,
-                const std::optional<std::string>& table_schema,
-                const std::shared_ptr<MemoryPool>& memory_pool,
-                const std::shared_ptr<Executor>& executor,
-                const std::shared_ptr<FileSystem>& specific_file_system,
-                const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
-                const std::shared_ptr<RealtimeContext>& realtime_context,
-                const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
-                const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
-                const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level);
+    ReadContext(
+        const std::string& path, const std::string& branch,
+        const std::vector<std::string>& read_field_names,
+        const std::vector<int32_t>& read_field_ids, const std::shared_ptr<Predicate>& predicate,
+        const std::shared_ptr<VectorSearch>& vector_search,
+        const std::shared_ptr<FullTextSearch>& full_text_search, bool enable_predicate_filter,
+        bool enable_prefetch, bool enable_late_materializing, uint32_t prefetch_batch_count,
+        uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
+        uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
+        const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor,
+        const std::shared_ptr<FileSystem>& specific_file_system,
+        const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
+        const std::shared_ptr<RealtimeContext>& realtime_context,
+        const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
+        const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
+        const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level,
+        const std::shared_ptr<Catalog>& catalog, const std::optional<Identifier>& identifier);
 
     ~ReadContext();
 
@@ -174,6 +173,16 @@ class PAIMON_EXPORT ReadContext {
         return warmup_level_;
     }
 
+    /// Returns the catalog supplying table metadata, or null if unset.
+    const std::shared_ptr<Catalog>& GetCatalog() const {
+        return catalog_;
+    }
+
+    /// Returns the table identifier supplied with the catalog.
+    const std::optional<Identifier>& GetIdentifier() const {
+        return identifier_;
+    }
+
     /// Whether a read schema (C ArrowSchema) for nested column pruning was provided.
     bool HasReadSchema() const {
         return read_schema_ != nullptr && read_schema_->release != nullptr;
@@ -217,6 +226,8 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<Cache> cache_;
     std::shared_ptr<FormatTable> format_table_;
     WarmupLevel warmup_level_;
+    std::shared_ptr<Catalog> catalog_;
+    std::optional<Identifier> identifier_;
     // Owns schema resources and releases ArrowSchema::release in destructor.
     std::unique_ptr<ArrowSchema> read_schema_;
 };

include/paimon/scan_context.h

@@ -58,7 +58,9 @@ class PAIMON_EXPORT ScanContext {
                 const std::optional<std::string>& table_schema,
                 const std::map<std::string, std::string>& options,
                 const std::shared_ptr<Cache>& cache,
-                const std::shared_ptr<FormatTable>& format_table);
+                const std::shared_ptr<FormatTable>& format_table,
+                const std::shared_ptr<Catalog>& catalog,
+                const std::optional<Identifier>& identifier);
 
     ~ScanContext();
 
@@ -115,6 +117,16 @@ class PAIMON_EXPORT ScanContext {
         return format_table_;
     }
 
+    /// Returns the catalog supplying table metadata, or null if unset.
+    const std::shared_ptr<Catalog>& GetCatalog() const {
+        return catalog_;
+    }
+
+    /// Returns the table identifier supplied with the catalog.
+    const std::optional<Identifier>& GetIdentifier() const {
+        return identifier_;
+    }
+
  private:
     std::string path_;
     bool is_streaming_mode_;
@@ -129,6 +141,8 @@ class PAIMON_EXPORT ScanContext {
     std::map<std::string, std::string> options_;
     std::shared_ptr<Cache> cache_;
     std::shared_ptr<FormatTable> format_table_;
+    std::shared_ptr<Catalog> catalog_;
+    std::optional<Identifier> identifier_;
 };
 
 /// Filter configuration for table scan operations

include/paimon/table/format/format_table.h

@@ -157,7 +157,7 @@ class PAIMON_EXPORT FormatTable {
 
     /// A name to identify this table.
     std::string Name() const {
-        return identifier_.GetTableName();
+        return identifier_.GetObjectName();
     }
 
     /// Full name of the table, database.tableName.

src/paimon/common/data/blob_view_struct_test.cpp

@@ -40,7 +40,7 @@ class BlobViewStructTest : public testing::Test {
 
 TEST_F(BlobViewStructTest, TestConstructorAndGetters) {
     ASSERT_EQ(view_struct_.GetIdentifier().GetDatabaseName(), "test_db");
-    ASSERT_EQ(view_struct_.GetIdentifier().GetTableName(), "test_table");
+    ASSERT_EQ(view_struct_.GetIdentifier().GetObjectName(), "test_table");
     ASSERT_EQ(view_struct_.FieldId(), 7);
     ASSERT_EQ(view_struct_.RowId(), 1024);
 }
@@ -53,7 +53,7 @@ TEST_F(BlobViewStructTest, TestSerializeDeserializeRoundTrip) {
     ASSERT_OK_AND_ASSIGN(auto restored,
                          BlobViewStruct::Deserialize(serialized->data(), serialized->size()));
     ASSERT_EQ(restored->GetIdentifier().GetDatabaseName(), "test_db");
-    ASSERT_EQ(restored->GetIdentifier().GetTableName(), "test_table");
+    ASSERT_EQ(restored->GetIdentifier().GetObjectName(), "test_table");
     ASSERT_EQ(restored->FieldId(), 7);
     ASSERT_EQ(restored->RowId(), 1024);
 }
@@ -78,7 +78,8 @@ TEST_F(BlobViewStructTest, TestToString) {
 }
 
 TEST_F(BlobViewStructTest, TestSerializeRejectsUnknownDatabase) {
-    BlobViewStruct view_struct(Identifier("test_table"), /*field_id=*/7, /*row_id=*/1024);
+    BlobViewStruct view_struct(Identifier(Identifier::kUnknownDatabase, "test_table"),
+                               /*field_id=*/7, /*row_id=*/1024);
     ASSERT_NOK_WITH_MSG(view_struct.Serialize(pool_),
                         "upstream table identifier must include database name");
 }

src/paimon/common/data/blob_view_test.cpp

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