feat(format-table): support reading and writing format tables

A format table is a directory of data files of one format with no
snapshots and no manifests, laid out like a standard Hive table. Only
`parquet` and `orc` are supported. `Catalog::GetFormatTable()` loads
one, and the new `FormatTableCatalog` interface lets a catalog serve
that call itself.

A caller holding a table path keeps the interface it uses for every
other table: `TableScan::Create()`, `TableRead::Create()`,
`FileStoreWrite::Create()` and `FileStoreCommit::Create()` recognise a
format table from the schema under the path and dispatch to it, the way
Java Paimon serves both kinds through one `ReadBuilder` and one
`BatchWriteBuilder`. `FormatTableScan`, `FormatTableRead`,
`FormatTableWrite` and `FormatTableCommit` stay public for a caller that
already holds a `FormatTable`. One `FormatTableLoader` answers the "is
this a format table" question for all four entry points and hands back
the schema it read, so the schema file is read once. Anything a context
carries that a format table cannot honour is refused by name rather than
quietly dropped.

The scan discovers partitions by descending the directory layout -
`key=value` by default, the bare value under
`format-table.partition-path-only-value` - and packs a partition's files
into splits of about `source.split.target-size`, each holding whole
files. The read projects the table's columns, pushes a predicate into
the file readers and applies it exactly on request, and rebuilds the
partition columns from the directory names, since the data files
themselves do not carry them. Partition completion, the predicate split
and the read-schema order all come from `FieldMappingReader`, and a data
file is opened through `DataFileReaderFactory`, which the managed table
path uses too, so prefetch and the caches a `ReadContext` carries apply
here as well.

The write is two-phase: it stages each file under
`_temporary/.tmp.<uuid>` beside where it will be published, the layout
Java's `RenamingTwoPhaseOutputStream` stages under, and the commit
renames them into place. An overwriting commit first clears the
partitions it writes to, or the static partition it was given. Options
are read through `CoreOptions`, so a default lives in one place.

A split and a commit message are both public structs a caller may have
built itself, so each is checked for path containment, visibility and
partition binding before anything is read, renamed or deleted. Neither
has a serialized form: a format table's plan has no cross-runtime
encoding, so `Split::Serialize()` and `CommitMessage::Serialize()`
refuse them and a plan is read within the process that made it.

- `Catalog::GetTable()` and `Table::Create()` refuse a format table, and
  every other table type this library does not implement, instead of
  opening it as a managed table and looking for snapshots it never had.
- Every catalog validates a new table's schema through one entry point,
  so none accepts a table another could not open;
  `RestCatalog::CreateTable()` runs it now too.
- `PartitionPathUtils::GeneratePartitionPath()` takes the value-only
  layout as a parameter and refuses a partition value that cannot name a
  directory. The managed table path goes through it too.
- `LazyConcatBatchReader` opens one file of a split at a time, names the
  file in every failure and keeps answering with the failure it stopped
  at, as `BatchReader` requires.

`docs/source/user_guide/format_table.rst` describes the layout, the
options, and what is still missing against Java, including
`dynamic-partition-overwrite` and partition filters beyond equality.

Author: SteNicholas

docs/source/api.rst

@@ -27,6 +27,7 @@ API Reference
    api/commit
    api/scan
    api/read
+   api/format_table
    api/predicate
    api/file_format
    api/file_system

docs/source/api/catalog.rst

@@ -27,6 +27,10 @@ Interface
 .. doxygenclass:: paimon::Catalog
    :members:
 
+.. doxygenclass:: paimon::FormatTableCatalog
+   :members:
+   :undoc-members:
+
 .. doxygenclass:: paimon::Identifier
    :members:
    :undoc-members:

docs/source/api/format_table.rst

@@ -0,0 +1,59 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements.  See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership.  The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied.  See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+============
+Format Table
+============
+
+.. _cpp-api-format-table:
+
+Table
+=====
+
+.. doxygenclass:: paimon::FormatTable
+   :members:
+   :undoc-members:
+
+Read
+====
+
+.. doxygenclass:: paimon::FormatTableScan
+   :members:
+   :undoc-members:
+
+.. doxygenstruct:: paimon::FormatDataSplit
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FormatTableRead
+   :members:
+   :undoc-members:
+
+Write
+=====
+
+.. doxygenclass:: paimon::FormatTableWrite
+   :members:
+   :undoc-members:
+
+.. doxygenstruct:: paimon::FormatCommitMessage
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FormatTableCommit
+   :members:
+   :undoc-members:

docs/source/user_guide.rst

@@ -33,6 +33,7 @@ User Guide
    user_guide/data_types
    user_guide/primary_key_table
    user_guide/append_only_table
+   user_guide/format_table
    user_guide/system_tables
    user_guide/write
    user_guide/commit

docs/source/user_guide/format_table.rst

@@ -20,23 +20,30 @@
 
 #pragma once
 
-#include "paimon/commit_context.h"           // IWYU pragma: export
-#include "paimon/defs.h"                     // IWYU pragma: export
-#include "paimon/factories/factory.h"        // IWYU pragma: export
-#include "paimon/file_store_commit.h"        // IWYU pragma: export
-#include "paimon/file_store_write.h"         // IWYU pragma: export
-#include "paimon/fs/file_system_factory.h"   // IWYU pragma: export
-#include "paimon/memory/memory_pool.h"       // IWYU pragma: export
-#include "paimon/predicate/predicate.h"      // IWYU pragma: export
-#include "paimon/read_context.h"             // IWYU pragma: export
-#include "paimon/reader/batch_reader.h"      // IWYU pragma: export
-#include "paimon/record_batch.h"             // IWYU pragma: export
-#include "paimon/result.h"                   // IWYU pragma: export
-#include "paimon/scan_context.h"             // IWYU pragma: export
-#include "paimon/status.h"                   // IWYU pragma: export
-#include "paimon/table/source/table_read.h"  // IWYU pragma: export
-#include "paimon/table/source/table_scan.h"  // IWYU pragma: export
-#include "paimon/write_context.h"            // IWYU pragma: export
+#include "paimon/commit_context.h"                      // IWYU pragma: export
+#include "paimon/defs.h"                                // IWYU pragma: export
+#include "paimon/factories/factory.h"                   // IWYU pragma: export
+#include "paimon/file_store_commit.h"                   // IWYU pragma: export
+#include "paimon/file_store_write.h"                    // IWYU pragma: export
+#include "paimon/fs/file_system_factory.h"              // IWYU pragma: export
+#include "paimon/memory/memory_pool.h"                  // IWYU pragma: export
+#include "paimon/predicate/predicate.h"                 // IWYU pragma: export
+#include "paimon/read_context.h"                        // IWYU pragma: export
+#include "paimon/reader/batch_reader.h"                 // IWYU pragma: export
+#include "paimon/record_batch.h"                        // IWYU pragma: export
+#include "paimon/result.h"                              // IWYU pragma: export
+#include "paimon/scan_context.h"                        // IWYU pragma: export
+#include "paimon/status.h"                              // IWYU pragma: export
+#include "paimon/table/format/format_commit_message.h"  // IWYU pragma: export
+#include "paimon/table/format/format_data_split.h"      // IWYU pragma: export
+#include "paimon/table/format/format_table.h"           // IWYU pragma: export
+#include "paimon/table/format/format_table_commit.h"    // IWYU pragma: export
+#include "paimon/table/format/format_table_read.h"      // IWYU pragma: export
+#include "paimon/table/format/format_table_scan.h"      // IWYU pragma: export
+#include "paimon/table/format/format_table_write.h"     // IWYU pragma: export
+#include "paimon/table/source/table_read.h"             // IWYU pragma: export
+#include "paimon/table/source/table_scan.h"             // IWYU pragma: export
+#include "paimon/write_context.h"                       // IWYU pragma: export
 
 // IWYU pragma: begin_exports
 #include "paimon/realtime/realtime_context.h"

include/paimon/api.h

@@ -40,6 +40,7 @@ namespace paimon {
 using Instant = std::variant<std::string, int64_t>;
 
 class Database;
+class FormatTable;
 class Table;
 class View;
 class Schema;
@@ -208,6 +209,23 @@ class PAIMON_EXPORT Catalog {
     ///         snapshot id ascending, or an error status.
     virtual Result<std::vector<SnapshotInfo>> ListSnapshots(
         const Identifier& identifier, const std::string& branch = "") const = 0;
+
+    /// Gets a format table: a directory of data files laid out like a standard Hive table.
+    ///
+    /// A format table is read and written through `FormatTableScan`, `FormatTableRead`,
+    /// `FormatTableWrite` and `FormatTableCommit` rather than through the managed table path, so
+    /// it is loaded through its own method instead of `GetTable()`.
+    ///
+    /// Not virtual, since `Catalog` is derived from outside this library: a catalog that can load
+    /// a format table itself derives from `FormatTableCatalog` as well, and this finds it with a
+    /// `dynamic_cast`. Otherwise it reads the location and the schema through the methods above,
+    /// which costs two requests that can disagree, and treats everything below the location as
+    /// data.
+    ///
+    /// @param identifier Identifier of the table to get.
+    /// @return A result containing the format table, or an error status if the table does not
+    /// exist or its `type` option is not `format-table`.
+    Result<std::shared_ptr<FormatTable>> GetFormatTable(const Identifier& identifier) const;
 };
 
 }  // namespace paimon

include/paimon/catalog/catalog.h

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+class FormatTable;
+class Identifier;
+
+/// What a catalog implements when it can load a format table itself.
+///
+/// It is a separate base rather than a virtual on `Catalog`, which is exported and derived from
+/// outside this library: a virtual added there would have no slot in an already-compiled subclass.
+/// A catalog inherits this alongside `Catalog`, and `Catalog::GetFormatTable()` finds it with a
+/// `dynamic_cast`.
+///
+/// A catalog that does not implement it still serves format tables: `Catalog::GetFormatTable()`
+/// falls back to reading the location and the schema through the methods every catalog has.
+class PAIMON_EXPORT FormatTableCatalog {
+ public:
+    virtual ~FormatTableCatalog() = default;
+
+    /// Loads `identifier` as a format table.
+    ///
+    /// The catalog decides both halves the fallback has to guess at: whether the location and the
+    /// schema can be read in one round trip, and whether it put this table's metadata under the
+    /// table path.
+    ///
+    /// @param identifier Identifier of the table to load.
+    /// @return A result containing the format table, or an error status if the table does not
+    /// exist or its `type` option is not `format-table`.
+    virtual Result<std::shared_ptr<FormatTable>> LoadFormatTable(
+        const Identifier& identifier) const = 0;
+};
+
+}  // namespace paimon

include/paimon/catalog/format_table_catalog.h

@@ -118,10 +118,33 @@ struct PAIMON_EXPORT Options {
     /// "page-size" - Memory page size, default value 64 kb.
     static const char PAGE_SIZE[];
 
+    /// "type" - Type of the table. Default value is "table" (a managed paimon table);
+    /// "format-table" declares a directory of plain data files laid out like a Hive table.
+    static const char TYPE[];
+
     /// "file.format" - Specify the message format of data files.
     /// Default value is parquet.
     static const char FILE_FORMAT[];
 
+    /// "format-table.file.compression" - File compression of a format table. It is consulted
+    /// after "file.compression" and before the bare "compression" key an engine's own writer
+    /// reads; when none of the three is set, the format decides.
+    static const char FORMAT_TABLE_FILE_COMPRESSION[];
+
+    /// "format-table.partition-path-only-value" - Whether a format table names a partition
+    /// directory by its value alone ("2025/01/") instead of "key=value"
+    /// ("year=2025/month=01/"). Default false.
+    static const char FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE[];
+
+    /// "metastore.partitioned-table" - Whether a table's partitions are registered with the
+    /// catalog, which then decides their visibility. Default false; paimon-cpp reads partitions
+    /// from the directory layout and rejects a table that sets it.
+    static const char METASTORE_PARTITIONED_TABLE[];
+
+    /// "file.suffix.include.compression" - Whether a data file's name carries the compression it
+    /// was written with, for a format that records it inside the file. Default false.
+    static const char FILE_SUFFIX_INCLUDE_COMPRESSION[];
+
     /// "file-system" - Specify the file system.
     /// Default value is local.
     static const char FILE_SYSTEM[];

include/paimon/defs.h

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <map>
+#include <string>
+
+#include "paimon/commit_message.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// One file a `FormatTableWrite` has written but not yet published.
+///
+/// The file is complete on disk under `temp_file_path`, which a scan skips; committing renames it
+/// to `file_path`, which is what makes it part of the table.
+///
+/// It is a `CommitMessage` so that a format table can be written and committed through
+/// `FileStoreWrite` and `FileStoreCommit` like any other table, as Java Paimon's
+/// `TwoPhaseCommitMessage` is. It names a staged path rather than files to record in a manifest,
+/// so `CommitMessage::Serialize()` refuses it: there is no cross-runtime encoding for one, and a
+/// write and its commit belong to the same process.
+struct PAIMON_EXPORT FormatCommitMessage : public CommitMessage {
+    FormatCommitMessage(const std::string& _temp_file_path, const std::string& _file_path,
+                        const std::map<std::string, std::string>& _partition, int64_t _record_count,
+                        int64_t _file_size)
+        : temp_file_path(_temp_file_path),
+          file_path(_file_path),
+          partition(_partition),
+          record_count(_record_count),
+          file_size(_file_size) {}
+
+    ~FormatCommitMessage() override = default;
+
+    std::string ToString() const;
+
+    /// Path the data was written to: a hidden file a scan skips.
+    std::string temp_file_path;
+    /// Path the file is renamed to when the write is committed.
+    std::string file_path;
+    /// Partition the file belongs to, empty when the table is not partitioned.
+    std::map<std::string, std::string> partition;
+    /// Rows written to the file.
+    int64_t record_count;
+    /// Size of the written file in bytes.
+    int64_t file_size;
+};
+
+}  // namespace paimon