Dev29

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,7 +53,9 @@ 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;
     Result<std::string> GetBranchNameOrDefault() const;
@@ -67,10 +69,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,8 +33,11 @@
 
 namespace paimon {
 
+class TableScan;
+class TableRead;
+
 /// A table provides basic abstraction for table type.
-class PAIMON_EXPORT Table {
+class PAIMON_EXPORT Table : public std::enable_shared_from_this<Table> {
  public:
     /// Creates a table by loading the latest schema from the specified table path.
     ///
@@ -47,19 +50,21 @@ class PAIMON_EXPORT Table {
                                                  const Identifier& identifier);
 
     /// @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<Schema>& schema, const Identifier& identifier,
+          const std::string& uuid = "")
+        : schema_(schema), identifier_(identifier), uuid_(uuid) {}
 
     ~Table() = default;
 
     /// 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 metastore UUID, falling back to `FullName()`.
     /// @note Use `CatalogUuid()` for commit requests to avoid sending a table name as an ID.
@@ -78,10 +83,42 @@ class PAIMON_EXPORT Table {
         return schema_;
     }
 
+    /// Creates a scan of this table from a finished scan context, standing in for
+    /// `TableScan::Create`. The scan is handed the table's shared owner, so it shares the
+    /// caller's hold on the table.
+    /// @param context The finished scan context.
+    /// @return The scan, or the error `TableScan::Create` reports. A table no `shared_ptr` holds
+    ///         cannot hand itself over and reports that.
+    Result<std::unique_ptr<TableScan>> NewScan(std::unique_ptr<ScanContext> context) const;
+
+    /// Creates a read of this table from a finished read context, standing in for
+    /// `TableRead::Create`. The read is handed the table's shared owner, so it shares the
+    /// caller's hold on the table.
+    /// @param context The finished read context.
+    /// @return The read, or the error `TableRead::Create` reports. A table no `shared_ptr` holds
+    ///         cannot hand itself over and reports that.
+    Result<std::unique_ptr<TableRead>> NewRead(std::unique_ptr<ReadContext> context) const;
+
+    /// Creates a write of this table from a finished write context, standing in for
+    /// `FileStoreWrite::Create`. The write is handed the table's shared owner, so it shares the
+    /// caller's hold on the table.
+    /// @param context The finished write context.
+    /// @return The write, or the error `FileStoreWrite::Create` reports. A table no `shared_ptr`
+    ///         holds cannot hand itself over and reports that.
+    Result<std::unique_ptr<FileStoreWrite>> NewWrite(std::unique_ptr<WriteContext> context) const;
+
+    /// Creates a commit of this table from a finished commit context, standing in for
+    /// `FileStoreCommit::Create`. The commit is handed the table's shared owner, so it shares
+    /// the caller's hold on the table.
+    /// @param context The finished commit context.
+    /// @return The commit, or the error `FileStoreCommit::Create` reports. A table no
+    ///         `shared_ptr` holds cannot hand itself over and reports that.
+    Result<std::unique_ptr<FileStoreCommit>> NewCommit(
+        std::unique_ptr<CommitContext> context) const;
+
  private:
     std::shared_ptr<Schema> schema_;
-    std::string database_;
-    std::string table_name_;
+    Identifier identifier_;
     std::string uuid_;
 };
 

include/paimon/file_store_commit.h

@@ -39,6 +39,7 @@
 namespace paimon {
 class CommitContext;
 class CommitMessage;
+class Table;
 
 /// Interface for commit operations in a file store.
 ///
@@ -57,6 +58,15 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @return A Result containing a unique pointer to the `FileStoreCommit` instance.
     static Result<std::unique_ptr<FileStoreCommit>> Create(std::unique_ptr<CommitContext> context);
 
+    /// Create an instance of `FileStoreCommit` for a table the caller already opened.
+    ///
+    /// @param context A unique pointer to the `CommitContext` used for commit operations.
+    /// @param table The opened table, or null when the caller has no table to hand over.
+    ///
+    /// @return A Result containing a unique pointer to the `FileStoreCommit` instance.
+    static Result<std::unique_ptr<FileStoreCommit>> Create(
+        std::unique_ptr<CommitContext> context, const std::shared_ptr<const Table>& table);
+
     virtual ~FileStoreCommit() = default;
 
     /// Commit changes to the file store.

include/paimon/file_store_write.h

@@ -37,6 +37,7 @@
 
 namespace paimon {
 class RecordBatch;
+class Table;
 class WriteContext;
 
 /// Interface for write operations in a file store.
@@ -49,6 +50,15 @@ class PAIMON_EXPORT FileStoreWrite {
     /// @return A Result containing a unique pointer to the `FileStoreWrite` instance.
     static Result<std::unique_ptr<FileStoreWrite>> Create(std::unique_ptr<WriteContext> context);
 
+    /// Create an instance of `FileStoreWrite` for a table the caller already opened.
+    ///
+    /// @param context A unique pointer to the `WriteContext` used for write operations.
+    /// @param table The opened table, or null when the caller has no table to hand over.
+    ///
+    /// @return A Result containing a unique pointer to the `FileStoreWrite` instance.
+    static Result<std::unique_ptr<FileStoreWrite>> Create(
+        std::unique_ptr<WriteContext> context, const std::shared_ptr<const Table>& table);
+
     virtual ~FileStoreWrite() = default;
 
     /// Support write an input `RecordBatch` to internal buffer or file.

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.

include/paimon/table/source/table_read.h

@@ -32,6 +32,7 @@
 
 namespace paimon {
 class ReadContext;
+class Table;
 
 /// Given a `Split` or a list of `Split`, generate a reader for batch reading.
 ///
@@ -48,6 +49,14 @@ class PAIMON_EXPORT TableRead {
     /// @return A Result containing a unique pointer to the `TableRead` instance.
     static Result<std::unique_ptr<TableRead>> Create(std::unique_ptr<ReadContext> context);
 
+    /// Create an instance of `TableRead` for a table the caller already opened.
+    ///
+    /// @param context A unique pointer to the `ReadContext` used for read operations.
+    /// @param table The opened table, or null when the caller has no table to hand over.
+    /// @return A Result containing a unique pointer to the `TableRead` instance.
+    static Result<std::unique_ptr<TableRead>> Create(std::unique_ptr<ReadContext> context,
+                                                     const std::shared_ptr<const Table>& table);
+
     /// Creates a `BatchReader` instance for reading data.
     ///
     /// This method creates a BatchReader that will be responsible for reading data from the

include/paimon/table/source/table_scan.h

@@ -32,6 +32,7 @@
 
 namespace paimon {
 class ScanContext;
+class Table;
 
 /// A scanner interface for reading table's meta and create a plan.
 class PAIMON_EXPORT TableScan {
@@ -42,6 +43,14 @@ class PAIMON_EXPORT TableScan {
     /// @return A Result containing a unique pointer to the `TableScan` instance.
     static Result<std::unique_ptr<TableScan>> Create(std::unique_ptr<ScanContext> context);
 
+    /// Create an instance of `TableScan` for a table the caller already opened.
+    ///
+    /// @param context A unique pointer to the `ScanContext` used for scan operations.
+    /// @param table The opened table, or null when the caller has no table to hand over.
+    /// @return A Result containing a unique pointer to the `TableScan` instance.
+    static Result<std::unique_ptr<TableScan>> Create(std::unique_ptr<ScanContext> context,
+                                                     const std::shared_ptr<const Table>& table);
+
     virtual ~TableScan() = default;
 
     /// Create a scan plan.

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");

src/paimon/core/catalog/file_system_catalog.cpp

@@ -103,7 +103,7 @@ Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const
 Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const {
     // Handle sys database global tables
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
-        return GlobalSystemTableLoader::IsSupported(identifier.GetTableName(), catalog_options_);
+        return GlobalSystemTableLoader::IsSupported(identifier.GetObjectName(), catalog_options_);
     }
     // The branch component is dropped when the data table identifier is rebuilt below, so the
     // identifier is validated as a whole here.
@@ -154,7 +154,7 @@ Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema*
             return Status::OK();
         } else {
             return Status::Invalid(
-                fmt::format("table {} already exist", identifier.GetTableName()));
+                fmt::format("table {} already exist", identifier.GetObjectName()));
         }
     }
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema,
@@ -276,7 +276,7 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
     // Handle sys database global tables
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
         PAIMON_ASSIGN_OR_RAISE(bool supported, GlobalSystemTableLoader::IsSupported(
-                                                   identifier.GetTableName(), catalog_options_));
+                                                   identifier.GetObjectName(), catalog_options_));
         if (!supported) {
             return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
         }
@@ -286,7 +286,7 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
         context.warehouse = warehouse_;
         context.catalog_options = catalog_options_;
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<SystemTable> system_table,
-                               GlobalSystemTableLoader::Load(identifier.GetTableName(), context));
+                               GlobalSystemTableLoader::Load(identifier.GetObjectName(), context));
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> arrow_schema,
                                system_table->ArrowSchema());
         return std::make_shared<SystemTableSchema>(std::move(arrow_schema));
@@ -333,8 +333,7 @@ Result<std::shared_ptr<Table>> FileSystemCatalog::GetTable(const Identifier& ide
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
     if (is_system_table) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, LoadTableSchema(identifier));
-        return std::make_shared<Table>(schema, identifier.GetDatabaseName(),
-                                       identifier.GetTableName());
+        return std::make_shared<Table>(schema, identifier);
     }
     PAIMON_ASSIGN_OR_RAISE(std::string table_path, GetTableLocation(identifier));
     // `Table::Create` reads the same schema file and rejects a format table there.