diff --git a/examples/read_write_demo.cpp b/examples/read_write_demo.cpp
index bcf7db1e1..df46fcc54 100644
--- a/examples/read_write_demo.cpp
+++ b/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,7 +119,7 @@ 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
@@ -121,7 +127,7 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
     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
@@ -129,7 +135,7 @@ paimon::Status Run(const std::string& root_path, const std::string& db_name,
     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;
diff --git a/include/paimon/catalog/identifier.h b/include/paimon/catalog/identifier.h
index bcc06208a..9766ee54e 100644
--- a/include/paimon/catalog/identifier.h
+++ b/include/paimon/catalog/identifier.h
@@ -27,7 +27,8 @@
 
 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[];
@@ -35,8 +36,7 @@ class PAIMON_EXPORT Identifier {
     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_;
diff --git a/include/paimon/catalog/table.h b/include/paimon/catalog/table.h
index ef68d1e44..bec5ed521 100644
--- a/include/paimon/catalog/table.h
+++ b/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_;
 };
 
diff --git a/include/paimon/file_store_commit.h b/include/paimon/file_store_commit.h
index 776ad52a8..556eddaab 100644
--- a/include/paimon/file_store_commit.h
+++ b/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.
diff --git a/include/paimon/file_store_write.h b/include/paimon/file_store_write.h
index 71ca910b6..286df6209 100644
--- a/include/paimon/file_store_write.h
+++ b/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.
diff --git a/include/paimon/table/format/format_table.h b/include/paimon/table/format/format_table.h
index 3f07880e4..a4c310254 100644
--- a/include/paimon/table/format/format_table.h
+++ b/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.
diff --git a/include/paimon/table/source/table_read.h b/include/paimon/table/source/table_read.h
index c77b323da..5e1c95de9 100644
--- a/include/paimon/table/source/table_read.h
+++ b/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
diff --git a/include/paimon/table/source/table_scan.h b/include/paimon/table/source/table_scan.h
index 8bae74c41..7b276ba48 100644
--- a/include/paimon/table/source/table_scan.h
+++ b/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.
diff --git a/src/paimon/common/data/blob_view_struct_test.cpp b/src/paimon/common/data/blob_view_struct_test.cpp
index 9668b7758..caa6cb9e1 100644
--- a/src/paimon/common/data/blob_view_struct_test.cpp
+++ b/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");
 }
diff --git a/src/paimon/common/data/blob_view_test.cpp b/src/paimon/common/data/blob_view_test.cpp
index 380bec0ec..92ff8fe3e 100644
--- a/src/paimon/common/data/blob_view_test.cpp
+++ b/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");
diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp
index 09be18d7d..e9bba9127 100644
--- a/src/paimon/core/catalog/file_system_catalog.cpp
+++ b/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.
diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp
index 74578e398..98607149d 100644
--- a/src/paimon/core/catalog/file_system_catalog_test.cpp
+++ b/src/paimon/core/catalog/file_system_catalog_test.cpp
@@ -546,7 +546,7 @@ TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
     ASSERT_EQ(table->FullName(), "db1.tbl1");
 
     ASSERT_NOK_WITH_MSG(catalog.GetTable(Identifier("db1", "table_xaxa")),
-                        "Identifier{database='db1', table='table_xaxa'} not exist");
+                        "Identifier{database='db1', object='table_xaxa'} not exist");
 
     ArrowSchemaRelease(&schema);
 }
@@ -759,7 +759,7 @@ TEST(FileSystemCatalogTest, TestValidateTableSchema) {
                                   /*ignore_if_exists=*/false));
 
     ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db0", "tbl0")),
-                        "Identifier{database=\'db0\', table=\'tbl0\'} not exist");
+                        "Identifier{database=\'db0\', object=\'tbl0\'} not exist");
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> table_schema, catalog.LoadTableSchema(identifier));
     auto data_schema = std::dynamic_pointer_cast<DataSchema>(table_schema);
     ASSERT_TRUE(data_schema != nullptr);
@@ -799,7 +799,7 @@ TEST(FileSystemCatalogTest, TestValidateTableSchema) {
 
     ASSERT_OK(fs->Delete(schema_path));
     ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(identifier),
-                        "Identifier{database=\'db1\', table=\'tbl1\'} not exist");
+                        "Identifier{database=\'db1\', object=\'tbl1\'} not exist");
 
     ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl$11")), "not exist");
     ArrowSchemaRelease(&schema);
@@ -879,7 +879,7 @@ TEST(FileSystemCatalogTest, TestDropTable) {
     ASSERT_NOK_WITH_MSG(
         catalog.DropTable(Identifier("test_db", "non_existent_tbl"),
                           /*ignore_if_not_exists=*/false),
-        "table Identifier{database='test_db', table='non_existent_tbl'} does not exist");
+        "table Identifier{database='test_db', object='non_existent_tbl'} does not exist");
 
     // Test 3: Drop existing table
     arrow::FieldVector fields = {
@@ -898,7 +898,7 @@ TEST(FileSystemCatalogTest, TestDropTable) {
     ASSERT_NOK_WITH_MSG(catalog.DropTable(Identifier("test_db", "tbl$system"),
                                           /*ignore_if_not_exists=*/false),
                         "Cannot 'dropTable' for system table 'Identifier{database='test_db', "
-                        "table='tbl$system'}', please use data table.");
+                        "object='tbl$system'}', please use data table.");
 
     ArrowSchemaRelease(&schema);
 }
@@ -923,7 +923,7 @@ TEST(FileSystemCatalogTest, TestRenameTable) {
         catalog.RenameTable(Identifier("test_db", "non_existent_tbl"),
                             Identifier("test_db", "new_tbl"),
                             /*ignore_if_not_exists=*/false),
-        "source table Identifier{database='test_db', table='non_existent_tbl'} does not exist");
+        "source table Identifier{database='test_db', object='non_existent_tbl'} does not exist");
 
     // Test 3: Normal rename
     arrow::FieldVector fields = {
@@ -950,7 +950,7 @@ TEST(FileSystemCatalogTest, TestRenameTable) {
     ASSERT_NOK_WITH_MSG(
         catalog.RenameTable(Identifier("test_db", "new_tbl"), Identifier("test_db", "tbl2"),
                             /*ignore_if_not_exists=*/false),
-        "target table Identifier{database='test_db', table='tbl2'} already exists");
+        "target table Identifier{database='test_db', object='tbl2'} already exists");
 
     // Test 5: Cross-database rename
     ASSERT_OK(catalog.CreateDatabase("test_db2", options, /*ignore_if_exists=*/false));
@@ -1553,7 +1553,7 @@ TEST(FileSystemCatalogTest, TestIdentifierNameValidationRules) {
         std::string table_error;
     };
     const std::vector<InvalidName> invalid_names = {
-        {"", "cannot be empty or whitespace", "Invalid table name"},
+        {"", "cannot be empty or whitespace", "Invalid object name"},
         {"   ", "cannot be empty or whitespace", "cannot be empty or whitespace"},
         {".", "cannot be '.' or '..'", "cannot be '.' or '..'"},
         {"..", "cannot be '.' or '..'", "cannot be '.' or '..'"},
diff --git a/src/paimon/core/catalog/identifier.cpp b/src/paimon/core/catalog/identifier.cpp
index 18fcc7975..69b2f555b 100644
--- a/src/paimon/core/catalog/identifier.cpp
+++ b/src/paimon/core/catalog/identifier.cpp
@@ -48,11 +48,8 @@ std::string BranchObjectName(const std::string& table, const std::string& branch
 
 }  // namespace
 
-Identifier::Identifier(const std::string& table)
-    : Identifier(std::string(kUnknownDatabase), table) {}
-
-Identifier::Identifier(const std::string& database, const std::string& table)
-    : database_(database), table_(table) {}
+Identifier::Identifier(const std::string& database, const std::string& object)
+    : database_(database), object_(object) {}
 
 Identifier::Identifier(const std::string& database, const std::string& table,
                        const std::string& branch)
@@ -62,24 +59,24 @@ bool Identifier::operator==(const Identifier& other) const {
     if (this == &other) {
         return true;
     }
-    return database_ == other.database_ && table_ == other.table_;
+    return database_ == other.database_ && object_ == other.object_;
 }
 
 const std::string& Identifier::GetDatabaseName() const {
     return database_;
 }
 
-const std::string& Identifier::GetTableName() const {
-    return table_;
+const std::string& Identifier::GetObjectName() const {
+    return object_;
 }
 
 Result<std::string> Identifier::GetDataTableName() const {
-    PAIMON_RETURN_NOT_OK(SplitTableName());
+    PAIMON_RETURN_NOT_OK(SplitObjectName());
     return data_table_;
 }
 
 Result<std::optional<std::string>> Identifier::GetBranchName() const {
-    PAIMON_RETURN_NOT_OK(SplitTableName());
+    PAIMON_RETURN_NOT_OK(SplitObjectName());
     return branch_;
 }
 
@@ -89,7 +86,7 @@ Result<std::string> Identifier::GetBranchNameOrDefault() const {
 }
 
 Result<std::optional<std::string>> Identifier::GetSystemTableName() const {
-    PAIMON_RETURN_NOT_OK(SplitTableName());
+    PAIMON_RETURN_NOT_OK(SplitObjectName());
     return system_table_;
 }
 
@@ -99,21 +96,21 @@ Result<bool> Identifier::IsSystemTable() const {
 }
 
 std::string Identifier::ToString() const {
-    return fmt::format("Identifier{{database='{}', table='{}'}}", database_, table_);
+    return fmt::format("Identifier{{database='{}', object='{}'}}", database_, object_);
 }
 
 int32_t Identifier::HashCode() const {
     int32_t hash = MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(database_.data()),
                                                     /*offset=*/0, database_.size());
-    return MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(table_.data()),
-                                            /*offset=*/0, table_.size(), hash);
+    return MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(object_.data()),
+                                            /*offset=*/0, object_.size(), hash);
 }
 
 std::string Identifier::GetFullName() const {
     if (database_ == kUnknownDatabase) {
-        return table_;
+        return object_;
     }
-    return fmt::format("{}.{}", database_, table_);
+    return fmt::format("{}.{}", database_, object_);
 }
 
 Result<Identifier> Identifier::FromString(const std::string& full_name) {
@@ -124,14 +121,14 @@ Result<Identifier> Identifier::FromString(const std::string& full_name) {
     const auto dot_pos = full_name.find('.');
     if (dot_pos == std::string::npos || dot_pos == 0 || dot_pos == full_name.size() - 1) {
         return Status::Invalid(
-            fmt::format("cannot get splits from '{}' to get database and table", full_name));
+            fmt::format("cannot get splits from '{}' to get database and object", full_name));
     }
     std::string database = full_name.substr(0, dot_pos);
-    std::string table = full_name.substr(dot_pos + 1);
-    return Identifier(database, table);
+    std::string object = full_name.substr(dot_pos + 1);
+    return Identifier(database, object);
 }
 
-Status Identifier::SplitTableName() const {
+Status Identifier::SplitObjectName() const {
     if (parsed_) {
         return Status::OK();
     }
@@ -139,9 +136,9 @@ Status Identifier::SplitTableName() const {
     std::optional<std::string> branch;
     std::optional<std::string> system_table;
     std::vector<std::string> splits =
-        StringUtils::Split(table_, kSystemTableSplitter, /*ignore_empty=*/false);
+        StringUtils::Split(object_, kSystemTableSplitter, /*ignore_empty=*/false);
     if (splits.size() == 1) {
-        data_table = table_;
+        data_table = object_;
     } else if (splits.size() == 2) {
         data_table = splits[0];
         if (StringUtils::StartsWith(splits[1], kSystemBranchPrefix, /*start_pos=*/0)) {
@@ -152,20 +149,20 @@ Status Identifier::SplitTableName() const {
     } else if (splits.size() == 3) {
         if (!StringUtils::StartsWith(splits[1], kSystemBranchPrefix, /*start_pos=*/0)) {
             return Status::Invalid(fmt::format(
-                "System table can only contain one '$' separator, but this is: {}", table_));
+                "System table can only contain one '$' separator, but this is: {}", object_));
         }
         data_table = splits[0];
         branch = splits[1].substr(std::strlen(kSystemBranchPrefix));
         system_table = splits[2];
     } else {
-        return Status::Invalid(fmt::format("Invalid table name: {}", table_));
+        return Status::Invalid(fmt::format("Invalid object name: {}", object_));
     }
     // A blank branch names the main branch everywhere else here, and the constructor above folds
     // it into the bare table name, so an object name spelling one was hand-built and would be
     // addressed as a branch of its own.
     if (data_table.empty() || (branch && StringUtils::IsNullOrWhitespaceOnly(branch.value())) ||
         (system_table && system_table->empty())) {
-        return Status::Invalid(fmt::format("Invalid table name: {}", table_));
+        return Status::Invalid(fmt::format("Invalid object name: {}", object_));
     }
     data_table_ = std::move(data_table);
     branch_ = std::move(branch);
diff --git a/src/paimon/core/catalog/identifier_test.cpp b/src/paimon/core/catalog/identifier_test.cpp
index d358a9255..f69683faa 100644
--- a/src/paimon/core/catalog/identifier_test.cpp
+++ b/src/paimon/core/catalog/identifier_test.cpp
@@ -26,13 +26,7 @@ namespace paimon::test {
 TEST(IdentifierTest, ConstructorAndGetters) {
     Identifier id("test_db", "test_table");
     ASSERT_EQ(id.GetDatabaseName(), "test_db");
-    ASSERT_EQ(id.GetTableName(), "test_table");
-}
-
-TEST(IdentifierTest, SingleArgumentConstructorUsesUnknownDatabase) {
-    Identifier id("test_table");
-    ASSERT_EQ(id.GetDatabaseName(), Identifier::kUnknownDatabase);
-    ASSERT_EQ(id.GetTableName(), "test_table");
+    ASSERT_EQ(id.GetObjectName(), "test_table");
 }
 
 TEST(IdentifierTest, EqualityOperator) {
@@ -46,18 +40,18 @@ TEST(IdentifierTest, EqualityOperator) {
 
 TEST(IdentifierTest, ToString) {
     Identifier id("my_db", "my_table");
-    ASSERT_EQ(id.ToString(), "Identifier{database='my_db', table='my_table'}");
+    ASSERT_EQ(id.ToString(), "Identifier{database='my_db', object='my_table'}");
 }
 
 TEST(IdentifierTest, EmptyDatabaseRemainsEmpty) {
     Identifier id("", "my_table");
     ASSERT_EQ(id.GetDatabaseName(), "");
-    ASSERT_EQ(id.GetTableName(), "my_table");
+    ASSERT_EQ(id.GetObjectName(), "my_table");
 }
 
 TEST(IdentifierTest, ParseDataTable) {
     Identifier id("db", "tbl");
-    ASSERT_EQ(id.GetTableName(), "tbl");
+    ASSERT_EQ(id.GetObjectName(), "tbl");
     ASSERT_OK_AND_ASSIGN(std::string data_table_name, id.GetDataTableName());
     ASSERT_EQ(data_table_name, "tbl");
     ASSERT_OK_AND_ASSIGN(std::optional<std::string> branch_name, id.GetBranchName());
@@ -72,7 +66,7 @@ TEST(IdentifierTest, ParseDataTable) {
 
 TEST(IdentifierTest, ParseSystemTable) {
     Identifier id("db", "tbl$options");
-    ASSERT_EQ(id.GetTableName(), "tbl$options");
+    ASSERT_EQ(id.GetObjectName(), "tbl$options");
     ASSERT_OK_AND_ASSIGN(std::string data_table_name, id.GetDataTableName());
     ASSERT_EQ(data_table_name, "tbl");
     ASSERT_OK_AND_ASSIGN(std::optional<std::string> branch_name, id.GetBranchName());
@@ -101,7 +95,7 @@ TEST(IdentifierTest, ParseBranchTable) {
 
 TEST(IdentifierTest, BranchConstructor) {
     Identifier branch_id("db", "tbl", "dev");
-    ASSERT_EQ(branch_id.GetTableName(), "tbl$branch_dev");
+    ASSERT_EQ(branch_id.GetObjectName(), "tbl$branch_dev");
     ASSERT_OK_AND_ASSIGN(std::string data_table_name, branch_id.GetDataTableName());
     ASSERT_EQ(data_table_name, "tbl");
     ASSERT_OK_AND_ASSIGN(std::optional<std::string> branch_name, branch_id.GetBranchName());
@@ -132,7 +126,7 @@ TEST(IdentifierTest, TestGetFullName) {
     {
         // test unknown database
         std::string table = "tbl$branch_dev$options";
-        Identifier id(table);
+        Identifier id(Identifier::kUnknownDatabase, table);
         ASSERT_EQ(table, id.GetFullName());
     }
     {
@@ -175,15 +169,15 @@ TEST(IdentifierTest, InvalidSystemTableName) {
                         "System table can only contain one '$' separator");
 
     Identifier too_many("db", "tbl$branch_dev$options$extra");
-    ASSERT_NOK_WITH_MSG(too_many.IsSystemTable(), "Invalid table name");
+    ASSERT_NOK_WITH_MSG(too_many.IsSystemTable(), "Invalid object name");
 }
 
 TEST(IdentifierTest, InvalidEmptySystemTableNameParts) {
-    ASSERT_NOK_WITH_MSG(Identifier("db", "$options").IsSystemTable(), "Invalid table name");
-    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$").IsSystemTable(), "Invalid table name");
-    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_").IsSystemTable(), "Invalid table name");
-    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_   ").IsSystemTable(), "Invalid table name");
-    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_dev$").IsSystemTable(), "Invalid table name");
+    ASSERT_NOK_WITH_MSG(Identifier("db", "$options").IsSystemTable(), "Invalid object name");
+    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$").IsSystemTable(), "Invalid object name");
+    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_").IsSystemTable(), "Invalid object name");
+    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_   ").IsSystemTable(), "Invalid object name");
+    ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$branch_dev$").IsSystemTable(), "Invalid object name");
     ASSERT_NOK_WITH_MSG(Identifier("db", "tbl$$options").IsSystemTable(),
                         "System table can only contain one '$' separator");
 }
diff --git a/src/paimon/core/operation/file_store_commit.cpp b/src/paimon/core/operation/file_store_commit.cpp
index c628a7ce9..9994f946e 100644
--- a/src/paimon/core/operation/file_store_commit.cpp
+++ b/src/paimon/core/operation/file_store_commit.cpp
@@ -174,6 +174,12 @@ Result<std::unique_ptr<FileStoreCommit>> NewFormatTableCommit(
 
 Result<std::unique_ptr<FileStoreCommit>> FileStoreCommit::Create(
     std::unique_ptr<CommitContext> ctx) {
+    return Create(std::move(ctx), nullptr);
+}
+
+Result<std::unique_ptr<FileStoreCommit>> FileStoreCommit::Create(
+    std::unique_ptr<CommitContext> ctx,
+    [[maybe_unused]] const std::shared_ptr<const Table>& table) {
     if (ctx == nullptr) {
         return Status::Invalid("commit context is null pointer");
     }
diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp
index 02ce10c45..05917ff76 100644
--- a/src/paimon/core/operation/file_store_commit_impl_test.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp
@@ -1105,7 +1105,7 @@ TEST_F(FileStoreCommitImplTest, TestCatalogCommitRetriesWhenRefused) {
         ASSERT_NE(failed.ToString().find(table_path_), std::string::npos) << failed.ToString();
         ASSERT_NE(failed.ToString().find("through catalog"), std::string::npos)
             << failed.ToString();
-        ASSERT_NE(failed.ToString().find("table='tbl'"), std::string::npos) << failed.ToString();
+        ASSERT_NE(failed.ToString().find("object='tbl'"), std::string::npos) << failed.ToString();
         ASSERT_NOK_WITH_MSG(failed, fmt::format("{} retries", max_retries));
 
         ASSERT_EQ(catalog->CommitCalls().size(), static_cast<size_t>(max_retries + 1));
@@ -1138,7 +1138,7 @@ TEST_F(FileStoreCommitImplTest, TestUnknownCatalogOutcomeKeepsTheManifests) {
     ASSERT_NOK_WITH_MSG(failed, "catalog unreachable");
     ASSERT_NE(failed.ToString().find(table_path_), std::string::npos) << failed.ToString();
     ASSERT_NE(failed.ToString().find("through catalog"), std::string::npos) << failed.ToString();
-    ASSERT_NE(failed.ToString().find("table='tbl'"), std::string::npos) << failed.ToString();
+    ASSERT_NE(failed.ToString().find("object='tbl'"), std::string::npos) << failed.ToString();
     ASSERT_NE(failed.ToString().find("snapshot uuid none"), std::string::npos) << failed.ToString();
 
     ASSERT_EQ(catalog->CommitCalls().size(), 1u);
diff --git a/src/paimon/core/operation/file_store_write.cpp b/src/paimon/core/operation/file_store_write.cpp
index 2ae664f8c..18bba4622 100644
--- a/src/paimon/core/operation/file_store_write.cpp
+++ b/src/paimon/core/operation/file_store_write.cpp
@@ -164,6 +164,11 @@ Status FileStoreWrite::RefreshCommittedSnapshot(int64_t) {
 }
 
 Result<std::unique_ptr<FileStoreWrite>> FileStoreWrite::Create(std::unique_ptr<WriteContext> ctx) {
+    return Create(std::move(ctx), nullptr);
+}
+
+Result<std::unique_ptr<FileStoreWrite>> FileStoreWrite::Create(
+    std::unique_ptr<WriteContext> ctx, [[maybe_unused]] const std::shared_ptr<const Table>& table) {
     if (ctx == nullptr) {
         return Status::Invalid("write context is null pointer");
     }
diff --git a/src/paimon/core/operation/read_context.cpp b/src/paimon/core/operation/read_context.cpp
index 1779d7300..eb8710e0d 100644
--- a/src/paimon/core/operation/read_context.cpp
+++ b/src/paimon/core/operation/read_context.cpp
@@ -70,12 +70,10 @@ std::string IdentifierToFlatJson(const Identifier& identifier) {
     auto& allocator = document.GetAllocator();
     document.AddMember(rapidjson::StringRef("database"),
                        rapidjson::Value(identifier.GetDatabaseName(), allocator), allocator);
-    // C++ `GetTableName()` returns the stored object name - branch and system-table suffix
-    // included - which is the Java `getObjectName()` this field mirrors, not Java's stripped
-    // `getTableName()` (that one is `GetDataTableName()` here). So a branch read reports the
-    // branch.
+    // `GetObjectName()` keeps any branch and system-table suffix, matching the Java
+    // `getObjectName()` this field mirrors. So a branch read reports the branch.
     document.AddMember(rapidjson::StringRef("object"),
-                       rapidjson::Value(identifier.GetTableName(), allocator), allocator);
+                       rapidjson::Value(identifier.GetObjectName(), allocator), allocator);
     // Compact, not the pretty form `RapidJsonUtil::ToJson` writes: the value goes into a header.
     rapidjson::StringBuffer buffer;
     rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
diff --git a/src/paimon/core/table/format/format_table_loader.cpp b/src/paimon/core/table/format/format_table_loader.cpp
index c50b85e4f..797bc3351 100644
--- a/src/paimon/core/table/format/format_table_loader.cpp
+++ b/src/paimon/core/table/format/format_table_loader.cpp
@@ -78,7 +78,8 @@ Result<std::shared_ptr<FormatTable>> FormatTableLoader::TryLoad(
     }
     // A context names a path, never a catalog identifier, so the table is known by the directory
     // it sits in.
-    return FormatTable::Create(file_system, table_path, Identifier(PathUtil::GetName(table_path)),
+    Identifier identifier(Identifier::kUnknownDatabase, PathUtil::GetName(table_path));
+    return FormatTable::Create(file_system, table_path, identifier,
                                checked_pointer_cast<DataSchema>(table_schema),
                                /*location_carries_paimon_metadata=*/true, options);
 }
diff --git a/src/paimon/core/table/source/table_read.cpp b/src/paimon/core/table/source/table_read.cpp
index 77e201206..66aa31c7f 100644
--- a/src/paimon/core/table/source/table_read.cpp
+++ b/src/paimon/core/table/source/table_read.cpp
@@ -170,6 +170,11 @@ Result<std::unique_ptr<TableRead>> NewFormatTableRead(const std::shared_ptr<Form
 }  // namespace
 
 Result<std::unique_ptr<TableRead>> TableRead::Create(std::unique_ptr<ReadContext> ctx) {
+    return Create(std::move(ctx), nullptr);
+}
+
+Result<std::unique_ptr<TableRead>> TableRead::Create(
+    std::unique_ptr<ReadContext> ctx, [[maybe_unused]] const std::shared_ptr<const Table>& table) {
     std::shared_ptr<ReadContext> context = std::move(ctx);
     if (context == nullptr) {
         return Status::Invalid("read context is null pointer");
diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp
index 7ede285fe..a77acb2d5 100644
--- a/src/paimon/core/table/source/table_scan.cpp
+++ b/src/paimon/core/table/source/table_scan.cpp
@@ -253,6 +253,12 @@ Result<std::vector<std::map<std::string, std::string>>> TableScan::ListPartition
 }
 
 Result<std::unique_ptr<TableScan>> TableScan::Create(std::unique_ptr<ScanContext> context) {
+    return Create(std::move(context), nullptr);
+}
+
+Result<std::unique_ptr<TableScan>> TableScan::Create(
+    std::unique_ptr<ScanContext> context,
+    [[maybe_unused]] const std::shared_ptr<const Table>& table) {
     if (context == nullptr) {
         return Status::Invalid("scan context is null pointer");
     }
diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp
index 2d67355a8..5a1a1de00 100644
--- a/src/paimon/core/table/system/system_table.cpp
+++ b/src/paimon/core/table/system/system_table.cpp
@@ -204,7 +204,8 @@ Result<std::optional<SystemTablePath>> SystemTableLoader::TryParsePath(const std
         return std::optional<SystemTablePath>(std::move(system_table_path));
     }
 
-    Identifier identifier(table_name);
+    // The database is unknown on a plain table path; only the object name is looked at below.
+    Identifier identifier(Identifier::kUnknownDatabase, table_name);
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
     if (!is_system_table) {
         return std::optional<SystemTablePath>();
diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp
index eaa574405..0faca368e 100644
--- a/src/paimon/core/table/table.cpp
+++ b/src/paimon/core/table/table.cpp
@@ -21,14 +21,36 @@
 #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 {
 
+namespace {
+
+/// The table's shared owner, which a `NewScan()`, `NewRead()`, `NewWrite()` or `NewCommit()` call
+/// hands to the operator it creates. A table no `shared_ptr` holds has no owner to hand over, and
+/// `shared_from_this()` would throw rather than say so, so the failure is stood up as a status.
+Result<std::shared_ptr<const Table>> SharedOwnerOf(const Table* table) {
+    if (std::shared_ptr<const Table> owner = table->weak_from_this().lock()) {
+        return owner;
+    }
+    return Status::Invalid("table is not held by a shared_ptr");
+}
+
+}  // namespace
+
 Result<std::shared_ptr<Table>> Table::Create(const std::shared_ptr<FileSystem>& file_system,
                                              const std::string& table_path,
                                              const Identifier& identifier) {
@@ -52,11 +74,29 @@ 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.GetTableName());
+    return std::make_shared<Table>(schema, identifier);
+}
+
+Result<std::unique_ptr<TableScan>> Table::NewScan(std::unique_ptr<ScanContext> context) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return TableScan::Create(std::move(context), table);
+}
+
+Result<std::unique_ptr<TableRead>> Table::NewRead(std::unique_ptr<ReadContext> context) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return TableRead::Create(std::move(context), table);
+}
+
+Result<std::unique_ptr<FileStoreWrite>> Table::NewWrite(
+    std::unique_ptr<WriteContext> context) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return FileStoreWrite::Create(std::move(context), table);
 }
 
-std::string Table::FullName() const {
-    return database_ == Identifier::kUnknownDatabase ? table_name_ : database_ + "." + table_name_;
+Result<std::unique_ptr<FileStoreCommit>> Table::NewCommit(
+    std::unique_ptr<CommitContext> context) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return FileStoreCommit::Create(std::move(context), table);
 }
 
 }  // namespace paimon
diff --git a/src/paimon/core/table/table_test.cpp b/src/paimon/core/table/table_test.cpp
index 60113ec8c..4e5c939ab 100644
--- a/src/paimon/core/table/table_test.cpp
+++ b/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);
@@ -51,7 +76,8 @@ TEST(TableTest, TestCreateWithUnknownDatabase) {
                          schema_manager.CreateTable(schema, partition_keys, primary_keys, options));
 
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table,
-                         Table::Create(dir->GetFileSystem(), dir->Str(), Identifier("tbl1")));
+                         Table::Create(dir->GetFileSystem(), dir->Str(),
+                                       Identifier(Identifier::kUnknownDatabase, "tbl1")));
 
     EXPECT_EQ(table->Name(), "tbl1");
     EXPECT_EQ(table->FullName(), "tbl1");
@@ -118,8 +144,109 @@ TEST(TableTest, TestCreateFailedWithNonExistSchema) {
     ASSERT_OK(fs->Delete(schema_path, /*recursive=*/false));
     // check create table failed
     ASSERT_NOK_WITH_MSG(
-        Table::Create(fs, dir->Str(), Identifier("tbl1")),
-        "load table schema for Identifier{database='unknown', table='tbl1'} failed");
+        Table::Create(fs, dir->Str(), Identifier(Identifier::kUnknownDatabase, "tbl1")),
+        "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");
+}
+
+TEST(TableTest, TestNewScanReadWriteCommitRequireTheTableToBeSharedOwned) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Table> table,
+        CreateManagedTable(dir->GetFileSystem(), dir->Str(), Identifier("db", "tbl")));
+
+    // An operator takes the table's shared owner from the table, and a stack object has none to
+    // take, so rather than crash reaching for one, the table reports it cannot hand itself over.
+    Table stack_table(table->LatestSchema(), Identifier("db", "tbl"));
+    ASSERT_NOK_WITH_MSG(stack_table.NewScan(nullptr), "table is not held by a shared_ptr");
+    ASSERT_NOK_WITH_MSG(stack_table.NewRead(nullptr), "table is not held by a shared_ptr");
+    ASSERT_NOK_WITH_MSG(stack_table.NewWrite(nullptr), "table is not held by a shared_ptr");
+    ASSERT_NOK_WITH_MSG(stack_table.NewCommit(nullptr), "table is not held by a shared_ptr");
 }
 
 }  // namespace paimon::test
diff --git a/src/paimon/core/utils/branch_manager_test.cpp b/src/paimon/core/utils/branch_manager_test.cpp
index 307a53f4b..c73459241 100644
--- a/src/paimon/core/utils/branch_manager_test.cpp
+++ b/src/paimon/core/utils/branch_manager_test.cpp
@@ -90,7 +90,7 @@ TEST(BranchManagerTest, TestResolveBranch) {
 
     ASSERT_NOK_WITH_MSG(BranchManager::ResolveBranch(Identifier("db", "tbl$a$b$c"), no_options,
                                                      std::nullopt, "commit"),
-                        "Invalid table name");
+                        "Invalid object name");
 
     ASSERT_NOK_WITH_MSG(
         BranchManager::ResolveBranch(Identifier("db", "tbl$branch_data"),
diff --git a/src/paimon/rest/rest_api.cpp b/src/paimon/rest/rest_api.cpp
index af5a78587..45865595d 100644
--- a/src/paimon/rest/rest_api.cpp
+++ b/src/paimon/rest/rest_api.cpp
@@ -273,11 +273,11 @@ Result<std::vector<std::string>> RestApi::ListTables(const std::string& database
 
 Result<GetTableResponse> RestApi::GetTable(const Identifier& identifier) const {
     return GetEntity<GetTableResponse>(
-        resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetTableName()), {});
+        resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetObjectName()), {});
 }
 
 Status RestApi::CreateTable(const Identifier& identifier, const std::string& schema_json) const {
-    CreateTableRequest request(identifier.GetDatabaseName(), identifier.GetTableName(),
+    CreateTableRequest request(identifier.GetDatabaseName(), identifier.GetObjectName(),
                                schema_json);
     PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString());
     return Execute("POST", resource_paths_.Tables(identifier.GetDatabaseName()), {}, body).status();
@@ -285,26 +285,26 @@ Status RestApi::CreateTable(const Identifier& identifier, const std::string& sch
 
 Status RestApi::DropTable(const Identifier& identifier) const {
     return Execute("DELETE",
-                   resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetTableName()),
+                   resource_paths_.Table(identifier.GetDatabaseName(), identifier.GetObjectName()),
                    {}, "")
         .status();
 }
 
 Status RestApi::RenameTable(const Identifier& from_table, const Identifier& to_table) const {
-    RenameTableRequest request(from_table.GetDatabaseName(), from_table.GetTableName(),
-                               to_table.GetDatabaseName(), to_table.GetTableName());
+    RenameTableRequest request(from_table.GetDatabaseName(), from_table.GetObjectName(),
+                               to_table.GetDatabaseName(), to_table.GetObjectName());
     PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString());
     return Execute("POST", resource_paths_.RenameTable(), {}, body).status();
 }
 
 Result<std::vector<Snapshot>> RestApi::ListSnapshots(const Identifier& identifier) const {
     return ListAllPages<ListSnapshotsResponse>(
-        resource_paths_.Snapshots(identifier.GetDatabaseName(), identifier.GetTableName()));
+        resource_paths_.Snapshots(identifier.GetDatabaseName(), identifier.GetObjectName()));
 }
 
 Result<std::optional<Snapshot>> RestApi::LoadSnapshot(const Identifier& identifier) const {
     std::string path =
-        resource_paths_.TableSnapshot(identifier.GetDatabaseName(), identifier.GetTableName());
+        resource_paths_.TableSnapshot(identifier.GetDatabaseName(), identifier.GetObjectName());
     Result<GetTableSnapshotResponse> response =
         GetEntity<GetTableSnapshotResponse>(path, /*query_params=*/{});
     if (response.ok()) {
@@ -329,7 +329,7 @@ Result<bool> RestApi::CommitSnapshot(const Identifier& identifier,
     CommitTableRequest request(table_uuid, base_snapshot_uuid, snapshot, statistics);
     PAIMON_ASSIGN_OR_RAISE(std::string body, request.ToJsonString());
     std::string path =
-        resource_paths_.CommitTable(identifier.GetDatabaseName(), identifier.GetTableName());
+        resource_paths_.CommitTable(identifier.GetDatabaseName(), identifier.GetObjectName());
     // Replaying an accepted commit can report a conflict and trigger manifest deletion.
     // Send once and let FilterAndCommit() recover an unknown outcome.
     PAIMON_ASSIGN_OR_RAISE(RestHttpClient::Response response,
@@ -341,7 +341,7 @@ Result<bool> RestApi::CommitSnapshot(const Identifier& identifier,
 
 Result<GetTableTokenResponse> RestApi::LoadTableToken(const Identifier& identifier) const {
     return GetEntity<GetTableTokenResponse>(
-        resource_paths_.TableToken(identifier.GetDatabaseName(), identifier.GetTableName()), {});
+        resource_paths_.TableToken(identifier.GetDatabaseName(), identifier.GetObjectName()), {});
 }
 
 }  // namespace paimon
diff --git a/src/paimon/rest/rest_catalog.cpp b/src/paimon/rest/rest_catalog.cpp
index acbc0cb78..8a13f3843 100644
--- a/src/paimon/rest/rest_catalog.cpp
+++ b/src/paimon/rest/rest_catalog.cpp
@@ -230,7 +230,7 @@ Status RestCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_sch
 
 Result<bool> RestCatalog::TableExists(const Identifier& identifier) const {
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
-        return GlobalSystemTableLoader::IsSupported(identifier.GetTableName(),
+        return GlobalSystemTableLoader::IsSupported(identifier.GetObjectName(),
                                                     api_->GetMergedOptions());
     }
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
@@ -375,7 +375,7 @@ Result<std::shared_ptr<Schema>> RestCatalog::LoadTableSchema(const Identifier& i
     // FileSystemCatalog.
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
         PAIMON_ASSIGN_OR_RAISE(bool supported,
-                               GlobalSystemTableLoader::IsSupported(identifier.GetTableName(),
+                               GlobalSystemTableLoader::IsSupported(identifier.GetObjectName(),
                                                                     api_->GetMergedOptions()));
         if (!supported) {
             return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
@@ -386,7 +386,7 @@ Result<std::shared_ptr<Schema>> RestCatalog::LoadTableSchema(const Identifier& i
         context.warehouse = warehouse_;
         context.catalog_options = api_->GetMergedOptions();
         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));
@@ -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.GetTableName(),
-                                   uuid);
+    return std::make_shared<Table>(schema, identifier, uuid);
 }
 
 std::string RestCatalog::GetRootPath() const {
@@ -494,7 +493,7 @@ Result<std::vector<SnapshotInfo>> RestCatalog::ListSnapshots(const Identifier& i
     PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "listSnapshots"));
     // `CheckNotBranch` and `CheckNotSystemTable` above leave the table name bare, so the branch
     // this is asked for is the only one the object name can carry.
-    Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetTableName(), branch);
+    Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetObjectName(), branch);
     // The Catalog interface has no pagination, so all pages are fetched; the server
     // does not order snapshots across pages while the contract requires ascending ids.
     PAIMON_ASSIGN_OR_RAISE(std::vector<Snapshot> snapshots, api_->ListSnapshots(load_identifier));
@@ -512,7 +511,7 @@ Result<std::optional<SnapshotInfo>> RestCatalog::GetLatestSnapshot(
     const Identifier& identifier, const std::string& branch) const {
     PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotBranch(identifier, "getLatestSnapshot"));
     PAIMON_RETURN_NOT_OK(CatalogUtils::CheckNotSystemTable(identifier, "getLatestSnapshot"));
-    Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetTableName(), branch);
+    Identifier load_identifier(identifier.GetDatabaseName(), identifier.GetObjectName(), branch);
     PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot, api_->LoadSnapshot(load_identifier));
     if (!snapshot) {
         return std::optional<SnapshotInfo>();
diff --git a/src/paimon/testing/mock/mock_catalog.h b/src/paimon/testing/mock/mock_catalog.h
index b9954e313..637730c5f 100644
--- a/src/paimon/testing/mock/mock_catalog.h
+++ b/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.GetTableName(), table_uuid_);
+        return std::make_shared<Table>(table_schema_, identifier, table_uuid_);
     }
 
     void SetTableSchema(const std::shared_ptr<Schema>& table_schema) {