refactor(table): introduce a common table hierarchy

docs/source/user_guide/catalog.rst

@@ -158,12 +158,27 @@ offsets when a real-time writer is created:
    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FileStoreWrite> writer,
                           paimon::FileStoreWrite::Create(std::move(write_context)));
 
-The ``Table`` handle is an equivalent entry point: ``table->NewCommit()``,
-``table->NewWrite()``, ``table->NewRead()`` and ``table->NewScan()`` each take the
-same finished context and return the same operator as ``FileStoreCommit::Create()``,
-``FileStoreWrite::Create()``, ``TableRead::Create()`` and ``TableScan::Create()``.
-Pick whichever reads better; the context still carries the catalog when one is
-needed. See the :ref:`Table API <cpp-api-catalog>`.
+``Catalog::GetTable()`` returns a common ``Table`` interface for managed, format and
+system tables. The managed implementation selects append-only or primary-key behavior
+from the loaded schema; system tables reject writes and commits.
+
+``table->NewCommit()``, ``NewWrite()``, ``NewRead()`` and ``NewScan()`` use the table's
+bound type, path, schema and file system. ``GetSchema()`` returns that schema without
+metadata I/O; reload through the catalog to refresh it. Contexts still provide execution
+settings, dynamic options and optional catalog services. Branch or type options that
+contradict a loaded managed table are rejected. The legacy static ``Create()`` methods
+load the table from their context before delegating to these methods.
+See the :ref:`Table API <cpp-api-catalog>`.
+
+An explicit ``WithBranch()`` or branch-qualified identifier participates in the same
+branch check. Omitting a branch allows the loaded table to supply it. The static
+scan and read factories use an explicit context branch first, then the branch in
+the schema options, and finally main.
+
+Global system tables under ``sys`` retain their catalog for deferred row loading.
+Before calling ``GetTable()`` for one, place the catalog in a ``std::shared_ptr``,
+as in the example above. ``LoadTableSchema()`` can inspect the schema without
+retaining the catalog.
 
 For real-time writes, also set ``WithStreamingMode(true)`` and
 ``WithRealtimeContext(...)`` with ``realtime.enabled=true``. After

docs/source/user_guide/format_table.rst

@@ -40,8 +40,10 @@ format table is an insert.
 
 Reading and writing
 -------------------
-A format table is not served through :cpp:func:`Catalog::GetTable`, which describes a managed
-table; use :cpp:func:`Catalog::GetFormatTable` instead.
+Use :cpp:func:`Catalog::GetTable` to load a format table through the common ``Table``
+interface. Its ``NewScan``, ``NewRead``, ``NewWrite`` and ``NewCommit`` methods use the
+loaded schema, location and file system. :cpp:func:`Catalog::GetFormatTable` also returns
+the concrete ``FormatTable`` interface for callers using the legacy context constructors.
 
 Reading and writing go through the entry points every other table uses: ``TableScan::Create``,
 ``TableRead::Create``, ``FileStoreWrite::Create`` and ``FileStoreCommit::Create``, each from its

include/paimon/catalog/catalog.h

@@ -48,7 +48,7 @@ class Tag;
 class Identifier;
 /// This interface is responsible for reading and writing metadata such as database/table from a
 /// paimon catalog.
-class PAIMON_EXPORT Catalog {
+class PAIMON_EXPORT Catalog : public std::enable_shared_from_this<Catalog> {
  public:
     static const char SYSTEM_DATABASE_NAME[];
     static const char SYSTEM_TABLE_SPLITTER[];
@@ -146,7 +146,7 @@ class PAIMON_EXPORT Catalog {
     virtual Status RenameTable(const Identifier& from_table, const Identifier& to_table,
                                bool ignore_if_not_exists) = 0;
 
-    /// Gets a table.
+    /// Gets a managed, format or read-only system table.
     ///
     /// @param identifier Identifier of the table to get.
     /// @return A result containing the table, or an error status.
@@ -249,11 +249,8 @@ class PAIMON_EXPORT Catalog {
 
     /// Gets a format table: a directory of data files laid out like a standard Hive table.
     ///
-    /// A format table carries no snapshots and no manifests, so it is loaded through its own
-    /// method rather than `GetTable()`. Reading and writing it go through the same `TableScan`,
-    /// `TableRead`, `FileStoreWrite` and `FileStoreCommit` entry points every other table uses:
-    /// hand the table back to `ScanContextBuilder`, `ReadContextBuilder`, `WriteContextBuilder` or
-    /// `CommitContextBuilder`, each of which takes one.
+    /// Returns the concrete format-table interface. `GetTable()` also loads format tables
+    /// through the common Table interface. Legacy context builders accept this typed handle.
     ///
     /// @param identifier Identifier of the table to get.
     /// @return A result containing the format table, or an error status if the table does not

include/paimon/catalog/table.h

@@ -33,50 +33,37 @@
 
 namespace paimon {
 
-class TableScan;
+class CommitContext;
+class FileStoreCommit;
+class FileStoreWrite;
+class ReadContext;
+class ScanContext;
 class TableRead;
+class TableScan;
+class WriteContext;
 
-/// A table provides basic abstraction for table type.
+/// Common interface for managed, format and system tables.
+/// Create operators with NewScan/NewRead/NewWrite/NewCommit; unsupported operations return errors.
+/// Operators use this table's type, path, schema and file system, with context execution settings.
 class PAIMON_EXPORT Table {
  public:
-    /// @param file_system File system the table's data is accessed through.
-    /// @param root_path Root path of the table in the file system.
-    /// @param uuid Metastore UUID, or empty if unavailable.
-    Table(const std::shared_ptr<FileSystem>& file_system, const std::string& root_path,
-          const std::shared_ptr<Schema>& schema, const Identifier& identifier,
-          const std::string& uuid = "");
-
-    ~Table() = default;
+    virtual ~Table() = default;
 
-    /// Returns the file system the table was opened with, which serves the table's
-    /// files. A catalog that hands out per-table file systems opens a table with the
-    /// one for that table; it is null when the opener had none.
+    /// Returns the table file system, or null if none was supplied.
     std::shared_ptr<FileSystem> GetFileSystem() const {
         return file_system_;
     }
 
-    /// Returns the root path of the table in its file system. It is empty when the
-    /// opener had no location for the table, which a mock in a test is one case of.
+    /// Returns the table root path, or an empty string if unavailable.
     const std::string& GetRootPath() const {
         return root_path_;
     }
 
-    /// A name to identify this table.
-    std::string Name() const {
-        return identifier_.GetObjectName();
-    }
-
-    /// Full name of the table, default is database.tableName.
-    std::string FullName() const {
-        return identifier_.GetFullName();
-    }
+    /// Returns the object name for data tables, or the system table name.
+    virtual std::string Name() const = 0;
 
-    /// Returns the branch the table was opened for, which its identifier carries; `main`
-    /// when it names none or a spelling of it, which its managers are bound to. A
-    /// malformed object name that hides the branch is reported.
-    Result<std::string> CurrentBranch() const {
-        return identifier_.GetBranchNameOrDefault();
-    }
+    /// Returns database.tableName for data tables, or Name() for system tables.
+    virtual std::string FullName() const = 0;
 
     /// Returns the metastore UUID, falling back to `FullName()`.
     /// @note Use `CatalogUuid()` for commit requests to avoid sending a table name as an ID.
@@ -90,42 +77,42 @@ class PAIMON_EXPORT Table {
         return uuid_.empty() ? std::nullopt : std::optional<std::string>(uuid_);
     }
 
-    /// Loads the latest schema of table.
-    std::shared_ptr<Schema> LatestSchema() const {
+    /// Returns the schema bound to this table instance without metadata I/O.
+    /// Reload through the catalog to obtain the latest schema; this accessor does not refresh it.
+    std::shared_ptr<Schema> GetSchema() const {
         return schema_;
     }
 
-    /// Creates a scan of this table from a finished scan context, the table-first entry point
-    /// that `TableScan::Create()` also builds through. The scan is derived from the context.
-    /// @param context The finished scan context.
-    /// @return The scan, or the error building it reports.
-    Result<std::unique_ptr<TableScan>> NewScan(std::unique_ptr<ScanContext> context) const;
-
-    /// Creates a read of this table from a finished read context, the table-first entry point
-    /// that `TableRead::Create()` also builds through. The read is derived from the context.
-    /// @param context The finished read context.
-    /// @return The read, or the error building it reports.
-    Result<std::unique_ptr<TableRead>> NewRead(std::unique_ptr<ReadContext> context) const;
-
-    /// Creates a write of this table from a finished write context, the table-first entry point
-    /// that `FileStoreWrite::Create()` also builds through. The write is derived from the context.
-    /// @param context The finished write context.
-    /// @return The write, or the error building it reports.
-    Result<std::unique_ptr<FileStoreWrite>> NewWrite(std::unique_ptr<WriteContext> context) const;
-
-    /// Creates a commit of this table from a finished commit context, the table-first entry point
-    /// that `FileStoreCommit::Create()` also builds through. The commit is derived from the
-    /// context.
-    /// @param context The finished commit context.
-    /// @return The commit, or the error building it reports.
-    Result<std::unique_ptr<FileStoreCommit>> NewCommit(
-        std::unique_ptr<CommitContext> context) const;
+    /// Creates a scan using this table's schema and options.
+    virtual Result<std::unique_ptr<TableScan>> NewScan(
+        std::unique_ptr<ScanContext> context) const = 0;
+
+    /// Creates a reader using this table's schema and options.
+    virtual Result<std::unique_ptr<TableRead>> NewRead(
+        std::unique_ptr<ReadContext> context) const = 0;
+
+    /// Creates a writer using this table's schema and options.
+    virtual Result<std::unique_ptr<FileStoreWrite>> NewWrite(
+        std::unique_ptr<WriteContext> context) const = 0;
+
+    /// Creates a commit using this table's schema and options.
+    virtual Result<std::unique_ptr<FileStoreCommit>> NewCommit(
+        std::unique_ptr<CommitContext> context) const = 0;
+
+ protected:
+    /// @param file_system File system the table's data is accessed through.
+    /// @param root_path Root path of the table in the file system.
+    /// @param uuid Metastore UUID, or empty if unavailable.
+    Table(const std::shared_ptr<FileSystem>& file_system, const std::string& root_path,
+          const std::shared_ptr<Schema>& schema, const Identifier& identifier,
+          const std::string& uuid);
 
- private:
     std::shared_ptr<FileSystem> file_system_;
     std::string root_path_;
     std::shared_ptr<Schema> schema_;
     Identifier identifier_;
+
+ private:
     std::string uuid_;
 };
 

include/paimon/table/format/format_table.h

@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "paimon/catalog/identifier.h"
+#include "paimon/catalog/table.h"
 #include "paimon/result.h"
 #include "paimon/schema/schema.h"
 #include "paimon/status.h"
@@ -53,7 +54,7 @@ class FileSystem;
 /// every context builder has, which is how a table whose schema lives in a metastore is reached.
 ///
 /// See `docs/source/user_guide/format_table.rst` for what is not supported yet.
-class PAIMON_EXPORT FormatTable {
+class PAIMON_EXPORT FormatTable : public Table {
  public:
     /// Formats a format table's files can be in.
     enum class Format {
@@ -115,11 +116,18 @@ class PAIMON_EXPORT FormatTable {
         const std::shared_ptr<FormatTable>& table,
         const std::map<std::string, std::string>& dynamic_options);
 
-    ~FormatTable();
+    ~FormatTable() override;
+
+    Result<std::unique_ptr<TableScan>> NewScan(std::unique_ptr<ScanContext> context) const override;
+    Result<std::unique_ptr<TableRead>> NewRead(std::unique_ptr<ReadContext> context) const override;
+    Result<std::unique_ptr<FileStoreWrite>> NewWrite(
+        std::unique_ptr<WriteContext> context) const override;
+    Result<std::unique_ptr<FileStoreCommit>> NewCommit(
+        std::unique_ptr<CommitContext> context) const override;
 
     /// Directory the data files live in.
     const std::string& Location() const {
-        return location_;
+        return root_path_;
     }
 
     /// Format of every data file in the directory.
@@ -156,16 +164,16 @@ class PAIMON_EXPORT FormatTable {
     }
 
     /// A name to identify this table.
-    std::string Name() const {
+    std::string Name() const override {
         return identifier_.GetObjectName();
     }
 
     /// Full name of the table, database.tableName.
-    std::string FullName() const;
+    std::string FullName() const override;
 
     /// Schema of the table, including its partition fields.
-    std::shared_ptr<DataSchema> LatestSchema() const {
-        return schema_;
+    std::shared_ptr<DataSchema> GetSchema() const {
+        return std::static_pointer_cast<DataSchema>(schema_);
     }
 
     /// Schema of the table as an arrow schema, including its partition fields.
@@ -186,16 +194,14 @@ class PAIMON_EXPORT FormatTable {
     }
 
  private:
+    Result<std::shared_ptr<FormatTable>> ForContext(
+        const std::map<std::string, std::string>& options) const;
     FormatTable(const std::shared_ptr<FileSystem>& file_system, const std::string& location,
                 const Identifier& identifier, const std::shared_ptr<DataSchema>& schema,
                 const std::map<std::string, std::string>& options, Format format,
                 const std::string& file_compression, const std::string& partition_default_name,
                 bool partition_only_value_in_path, bool location_carries_paimon_metadata);
 
-    std::shared_ptr<FileSystem> file_system_;
-    std::string location_;
-    Identifier identifier_;
-    std::shared_ptr<DataSchema> schema_;
     std::map<std::string, std::string> options_;
     Format format_;
     std::string file_compression_;

src/paimon/CMakeLists.txt

@@ -450,6 +450,10 @@ set(PAIMON_CORE_SRCS
     core/stats/simple_stats_converter.cpp
     core/stats/simple_stats.cpp
     core/stats/simple_stats_evolution.cpp
+    core/table/append_only_file_store_table.cpp
+    core/table/primary_key_file_store_table.cpp
+    core/table/file_store_table.cpp
+    core/table/table_factory.cpp
     core/table/file_store.cpp
     core/table/table.cpp
     core/table/bucket_mode.cpp

src/paimon/core/catalog/catalog_utils.cpp

@@ -26,6 +26,7 @@
 #include "fmt/format.h"
 #include "paimon/catalog/catalog.h"
 #include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/core/options/table_type.h"
 #include "paimon/core/utils/branch_manager.h"
 #include "paimon/defs.h"
@@ -149,4 +150,15 @@ Result<std::shared_ptr<FormatTable>> CatalogUtils::LoadFormatTableInTwoRequests(
                                location_carries_paimon_metadata, /*dynamic_options=*/{});
 }
 
+Identifier CatalogUtils::IdentifierFromPath(const std::string& table_path,
+                                            const std::string& branch) {
+    const std::string name = PathUtil::GetName(table_path);
+    const std::string parent = PathUtil::GetName(PathUtil::GetParentDirPath(table_path));
+    const std::string suffix = Catalog::DB_SUFFIX;
+    if (StringUtils::EndsWith(parent, suffix)) {
+        return Identifier(parent.substr(0, parent.size() - suffix.size()), name, branch);
+    }
+    return Identifier(Identifier::kUnknownDatabase, name, branch);
+}
+
 }  // namespace paimon

src/paimon/core/catalog/catalog_utils.h

@@ -38,6 +38,8 @@ class CatalogUtils {
     CatalogUtils() = delete;
     ~CatalogUtils() = delete;
 
+    /// Derives an identifier for a table loaded directly from a path.
+    static Identifier IdentifierFromPath(const std::string& path, const std::string& branch);
     /// Returns whether `db_name` is the reserved system database "sys".
     static bool IsSystemDatabase(const std::string& db_name);