fix

Author: lucasfang

include/paimon/catalog/table.h

@@ -33,6 +33,9 @@
 
 namespace paimon {
 
+class TableScan;
+class TableRead;
+
 /// A table provides basic abstraction for table type.
 class PAIMON_EXPORT Table {
  public:
@@ -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,34 @@ class PAIMON_EXPORT Table {
         return schema_;
     }
 
+    /// Creates a scan of this table from a finished scan context, standing in for
+    /// `TableScan::Create`.
+    /// @param context The finished scan context.
+    /// @return The scan, or the error `TableScan::Create` reports.
+    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`.
+    /// @param context The finished read context.
+    /// @return The read, or the error `TableRead::Create` reports.
+    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`.
+    /// @param context The finished write context.
+    /// @return The write, or the error `FileStoreWrite::Create` reports.
+    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`.
+    /// @param context The finished commit context.
+    /// @return The commit, or the error `FileStoreCommit::Create` reports.
+    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_;
 };
 

src/paimon/core/catalog/file_system_catalog.cpp

@@ -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.GetObjectName());
+        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.

src/paimon/core/table/table.cpp

@@ -21,11 +21,19 @@
 #include <optional>
 
 #include "fmt/format.h"
+#include "paimon/commit_context.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/core/catalog/catalog_utils.h"
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
+#include "paimon/file_store_commit.h"
+#include "paimon/file_store_write.h"
 #include "paimon/fs/file_system.h"
+#include "paimon/read_context.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/table_read.h"
+#include "paimon/table/source/table_scan.h"
+#include "paimon/write_context.h"
 
 namespace paimon {
 
@@ -52,12 +60,25 @@ Result<std::shared_ptr<Table>> Table::Create(const std::shared_ptr<FileSystem>&
         CatalogUtils::CheckManagedTableType(identifier, *latest_schema, "Table::Create"));
 
     auto schema = checked_pointer_cast<Schema>(*latest_schema);
-    return std::make_shared<Table>(schema, identifier.GetDatabaseName(),
-                                   identifier.GetObjectName());
+    return std::make_shared<Table>(schema, identifier);
 }
 
-std::string Table::FullName() const {
-    return database_ == Identifier::kUnknownDatabase ? table_name_ : database_ + "." + table_name_;
+Result<std::unique_ptr<TableScan>> Table::NewScan(std::unique_ptr<ScanContext> context) const {
+    return TableScan::Create(std::move(context));
+}
+
+Result<std::unique_ptr<TableRead>> Table::NewRead(std::unique_ptr<ReadContext> context) const {
+    return TableRead::Create(std::move(context));
+}
+
+Result<std::unique_ptr<FileStoreWrite>> Table::NewWrite(
+    std::unique_ptr<WriteContext> context) const {
+    return FileStoreWrite::Create(std::move(context));
+}
+
+Result<std::unique_ptr<FileStoreCommit>> Table::NewCommit(
+    std::unique_ptr<CommitContext> context) const {
+    return FileStoreCommit::Create(std::move(context));
 }
 
 }  // namespace paimon

src/paimon/core/table/table_test.cpp

@@ -25,13 +25,38 @@
 
 #include "arrow/api.h"
 #include "gtest/gtest.h"
+#include "paimon/commit_context.h"
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
+#include "paimon/file_store_commit.h"
+#include "paimon/file_store_write.h"
+#include "paimon/read_context.h"
+#include "paimon/scan_context.h"
 #include "paimon/schema/schema.h"
+#include "paimon/table/source/table_read.h"
+#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/mock/mock_catalog.h"
 #include "paimon/testing/utils/testharness.h"
+#include "paimon/write_context.h"
 
 namespace paimon::test {
 
+namespace {
+
+// Creates a managed table directory and opens it as a `Table` under `identifier`.
+Result<std::shared_ptr<Table>> CreateManagedTable(const std::shared_ptr<FileSystem>& file_system,
+                                                  const std::string& path,
+                                                  const Identifier& identifier) {
+    SchemaManager schema_manager(file_system, path);
+    auto schema = arrow::schema({arrow::field("id", arrow::int32(), /*nullable=*/false)});
+    PAIMON_ASSIGN_OR_RAISE([[maybe_unused]] std::unique_ptr<TableSchema> created,
+                           schema_manager.CreateTable(schema, /*partition_keys=*/{},
+                                                      /*primary_keys=*/{}, /*options=*/{}));
+    return Table::Create(file_system, path, identifier);
+}
+
+}  // namespace
+
 TEST(TableTest, TestCreateWithUnknownDatabase) {
     auto dir = UniqueTestDirectory::Create();
     ASSERT_TRUE(dir);
@@ -123,4 +148,89 @@ TEST(TableTest, TestCreateFailedWithNonExistSchema) {
         "load table schema for Identifier{database='unknown', object='tbl1'} failed");
 }
 
+TEST(TableTest, TestNewScanCreatesATableScan) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const Identifier identifier("db", "tbl");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
+                         CreateManagedTable(dir->GetFileSystem(), dir->Str(), identifier));
+
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetFileSystem(dir->GetFileSystem());
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> ctx,
+                         ScanContextBuilder(dir->Str()).WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<TableScan> scan,
+                         table->NewScan(std::move(ctx)));
+}
+
+TEST(TableTest, TestNewReadCreatesATableRead) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const Identifier identifier("db", "tbl");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
+                         CreateManagedTable(dir->GetFileSystem(), dir->Str(), identifier));
+
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetFileSystem(dir->GetFileSystem());
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> ctx,
+                         ReadContextBuilder(dir->Str()).WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<TableRead> read,
+                         table->NewRead(std::move(ctx)));
+}
+
+TEST(TableTest, TestNewWriteCreatesAFileStoreWrite) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const Identifier identifier("db", "tbl");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
+                         CreateManagedTable(dir->GetFileSystem(), dir->Str(), identifier));
+
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetFileSystem(dir->GetFileSystem());
+    // A catalog write reads the current schema from the catalog on the main branch.
+    catalog->SetTableSchema(table->LatestSchema());
+
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<WriteContext> ctx,
+        WriteContextBuilder(dir->Str(), "commit-user").WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<FileStoreWrite> write,
+                         table->NewWrite(std::move(ctx)));
+}
+
+TEST(TableTest, TestNewCommitCreatesAFileStoreCommit) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const Identifier identifier("db", "tbl");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
+                         CreateManagedTable(dir->GetFileSystem(), dir->Str(), identifier));
+
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetFileSystem(dir->GetFileSystem());
+    // Like a write, a catalog commit reads the current schema from the catalog.
+    catalog->SetTableSchema(table->LatestSchema());
+
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<CommitContext> ctx,
+        CommitContextBuilder(dir->Str(), "commit-user").WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<FileStoreCommit> commit,
+                         table->NewCommit(std::move(ctx)));
+}
+
+TEST(TableTest, TestNewScanReadWriteCommitRejectANullContext) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Table> table,
+        CreateManagedTable(dir->GetFileSystem(), dir->Str(), Identifier("db", "tbl")));
+
+    // The operators report the null context themselves; the table stands in without swallowing
+    // the error, so behavior matches `TableScan::Create` and friends.
+    ASSERT_NOK_WITH_MSG(table->NewScan(nullptr), "scan context is null pointer");
+    ASSERT_NOK_WITH_MSG(table->NewRead(nullptr), "read context is null pointer");
+    ASSERT_NOK_WITH_MSG(table->NewWrite(nullptr), "write context is null pointer");
+    ASSERT_NOK_WITH_MSG(table->NewCommit(nullptr), "commit context is null pointer");
+}
+
 }  // namespace paimon::test

src/paimon/rest/rest_catalog.cpp

@@ -448,8 +448,7 @@ Result<std::shared_ptr<Table>> RestCatalog::GetTable(const Identifier& identifie
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, LoadTableSchema(identifier, &uuid));
     PAIMON_RETURN_NOT_OK(
         CatalogUtils::CheckManagedTableType(identifier, schema, "Catalog::GetTable"));
-    return std::make_shared<Table>(schema, identifier.GetDatabaseName(), identifier.GetObjectName(),
-                                   uuid);
+    return std::make_shared<Table>(schema, identifier, uuid);
 }
 
 std::string RestCatalog::GetRootPath() const {

src/paimon/testing/mock/mock_catalog.h

@@ -95,8 +95,7 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog {
 
     Result<std::shared_ptr<Table>> GetTable(const Identifier& identifier) const override {
         ++get_table_calls_;
-        return std::make_shared<Table>(table_schema_, identifier.GetDatabaseName(),
-                                       identifier.GetObjectName(), table_uuid_);
+        return std::make_shared<Table>(table_schema_, identifier, table_uuid_);
     }
 
     void SetTableSchema(const std::shared_ptr<Schema>& table_schema) {