fix

Author: lucasfang

include/paimon/catalog/table.h

@@ -37,7 +37,7 @@ 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.
     ///
@@ -84,27 +84,35 @@ class PAIMON_EXPORT Table {
     }
 
     /// Creates a scan of this table from a finished scan context, standing in for
-    /// `TableScan::Create`.
+    /// `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.
+    /// @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`.
+    /// `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.
+    /// @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`.
+    /// `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.
+    /// @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`.
+    /// `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.
+    /// @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;
 

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

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

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

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

src/paimon/core/table/table.cpp

@@ -37,6 +37,20 @@
 
 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) {
@@ -64,21 +78,25 @@ Result<std::shared_ptr<Table>> Table::Create(const std::shared_ptr<FileSystem>&
 }
 
 Result<std::unique_ptr<TableScan>> Table::NewScan(std::unique_ptr<ScanContext> context) const {
-    return TableScan::Create(std::move(context));
+    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 {
-    return TableRead::Create(std::move(context));
+    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 {
-    return FileStoreWrite::Create(std::move(context));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return FileStoreWrite::Create(std::move(context), table);
 }
 
 Result<std::unique_ptr<FileStoreCommit>> Table::NewCommit(
     std::unique_ptr<CommitContext> context) const {
-    return FileStoreCommit::Create(std::move(context));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const Table> table, SharedOwnerOf(this));
+    return FileStoreCommit::Create(std::move(context), table);
 }
 
 }  // namespace paimon