refactor(format-table): serve format tables through the generic entry points

`TableScan`, `TableRead`, `FileStoreWrite` and `FileStoreCommit` now
recognise a format table from the schema under the table path and
dispatch to it, so a caller holding a path uses the same interface it
uses for every other table, the way Java Paimon serves both through one
`ReadBuilder` and one `BatchWriteBuilder`. The `FormatTable*` classes
stay public for a caller that already holds a `FormatTable`. One
`FormatTableLoader` answers the "is this a format table" question for
all four and hands back the schema it read, so the managed path does
not read the same file a second time. Anything a context carries that a
format table cannot honour is refused by name rather than dropped.

A read opens its data files through `DataFileReaderFactory`, the
component the managed table path opens its own with, so prefetch, the
read-ahead cache and the cache a `ReadContext` carries all apply here
too. Partition columns are completed by `FieldMappingReader` and the row
limit is left to the caller, which retires
`PartitionCompletingBatchReader` and `LimitBatchReader`. Batches carry
the leading `_VALUE_KIND` field every `BatchReader` promises, filled
with inserts, so `BatchReader`'s contract is unchanged after all and an
engine reading by field index sees one layout for every table.

Options are read through `CoreOptions`, so a default lives in one place:
`file.suffix.include.compression`,
`format-table.partition-path-only-value`,
`metastore.partitioned-table` and the format table compression chain
join it there. A write stages its files under `_temporary/.tmp.<uuid>`
beside where they will be published, the layout Java's
`RenamingTwoPhaseOutputStream` stages under.

- `FormatDataSplit` and `FormatCommitMessage` lose their 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.
- `LazyConcatBatchReader` keeps answering with the failure it stopped
  at, as `BatchReader` requires, rather than moving on to the next file.
- Schema validation runs one path for a `TableSchema` and a
  `DataSchema`, and `Table::Create()` reuses
  `CatalogUtils::CheckManagedTableType()`.
- `FormatFileListing` and `FormatPathValidation` are utility classes
  rather than free functions, `FormatFileNaming` takes the data file
  prefix rather than defaulting it, and the constants all three carry
  follow the `kName` convention.

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

Author: SteNicholas

docs/source/api/format_table.rst

@@ -35,7 +35,7 @@ Read
    :members:
    :undoc-members:
 
-.. doxygenclass:: paimon::FormatDataSplit
+.. doxygenstruct:: paimon::FormatDataSplit
    :members:
    :undoc-members:
 
@@ -50,7 +50,7 @@ Write
    :members:
    :undoc-members:
 
-.. doxygenclass:: paimon::FormatCommitMessage
+.. doxygenstruct:: paimon::FormatCommitMessage
    :members:
    :undoc-members:
 

docs/source/user_guide/format_table.rst

@@ -33,18 +33,86 @@ A partitioned format table uses the standard Hive directory layout, and its part
 discovered from that layout rather than from metadata. By default a partition directory is named
 ``key=value``; setting ``format-table.partition-path-only-value`` names it by the value alone.
 
-Because a directory of plain files records no row identity, a format table only accepts inserts,
-and reads return the table's own columns with no ``_VALUE_KIND`` field.
+Because a directory of plain files records no row identity, a format table only accepts inserts.
+Reads still carry the leading ``_VALUE_KIND`` field every ``BatchReader`` promises, so an engine
+that reads a batch by field index sees the same layout it does for a managed table; every row of a
+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. It is then read and written through
-``FormatTableScan``, ``FormatTableRead``, ``FormatTableWrite`` and ``FormatTableCommit`` rather
-than through the managed table path.
+table; use :cpp:func:`Catalog::GetFormatTable` instead.
 
-A write is two-phase, since a directory has no metadata to switch atomically: written files are
-staged under hidden names that a scan skips, and only the commit renames them into place.
+Reading and writing take either of two routes. The generic entry points read the table's schema -
+from under the table path, or from the one the context carries - and dispatch to a format table
+when its ``type`` says so, so a caller holding a table path uses the same interface it uses for
+every other table - ``TableScan::Create``, ``TableRead::Create``, ``FileStoreWrite::Create``
+and ``FileStoreCommit::Create``, each from its usual context builder. That is what Java Paimon does
+through ``FormatTable.newReadBuilder()`` and ``newBatchWriteBuilder()``. The format-specific classes
+``FormatTableScan``, ``FormatTableRead``, ``FormatTableWrite`` and ``FormatTableCommit`` remain
+public and take a ``FormatTable`` directly, for a caller that already has one and wants the narrower
+interface; Java keeps its own equivalents public for the same reason.
+
+Not all of the generic interfaces fit. ``FileStoreCommit`` is mostly about snapshots and manifests -
+expiring them, rolling back to one, filtering by a commit identifier recorded in one - and a format
+table keeps none of that state, so those calls are refused rather than quietly doing nothing.
+``FileStoreWrite::Compact()`` is refused for the same reason, and both write and commit take batch
+writes only, since there is no snapshot to record a commit identifier or a watermark in.
+``TableScan`` takes a partition filter and a limit; a predicate or a bucket filter is refused. Java
+refuses a predicate from ``FormatTableScan.withFilter`` too, but ``FormatReadBuilder.newScan()``
+splits one first and hands the partition half to the scan, so there a predicate over partition
+columns still prunes directories. See the limits below.
+
+Options given at the call win over the ones the schema stored, as they do for every other table -
+except ``type``, which is structural and is read from the schema alone, so one read or write
+cannot decide what kind of table this is. The merged result is validated, not the schema's own
+options, so an option a format table refuses - ``metastore.partitioned-table``, a file format
+nothing here can read - is refused wherever it comes from rather than dropped in silence.
+
+A setting the format path cannot act on is refused by name rather than quietly dropped:
+
+* ``ReadContextBuilder::SetReadSchema()``. A projected read schema can rename a column, prune a
+  nested one and give it metadata of its own; a format table's projection is a list of top-level
+  names, so name the columns instead.
+* ``WithStreamingMode()`` on a scan or a write, a global index result on a scan, and a real-time
+  context on a scan, a read or a write: a format table has no snapshots, no real-time store and no
+  index.
+* ``WriteContextBuilder::WithWriteSchema()``, which names a subset of the columns to write.
+* ``WriteContextBuilder::WithWriteId()``, which prefixes a postpone-bucket writer's files so one
+  compaction reader can put them back in order; a format table has no buckets.
+* ``CommitContextBuilder::IgnoreEmptyCommit(false)``, ``UseRESTCatalogCommit(true)`` and
+  ``AppendCommitCheckConflict(true)``. Keeping an empty commit means writing a snapshot that adds
+  no files, a rest-catalog commit sends that snapshot to a catalog, and the conflict check reads
+  the manifests of concurrent commits - none of which exist here. Each is refused only when set
+  away from its default, so an ordinary commit is unaffected.
+* A scan predicate or bucket filter, as above, and more than one partition filter: a scan descends
+  one directory layout, so it takes the values of a single partition rather than a set of them.
+
+What a data file is opened with is not one of the refusals. ``EnablePrefetch()``, the read-ahead
+cache and its ``CacheConfig``, and the ``Cache`` a read carries all apply, because a format table
+opens its files through the same component the managed table path opens its own with. What differs
+between the two paths is which files there are and how a row is put back together, not how a file
+is read.
+
+Some settings are not refused because they describe machinery the format path never reaches, and
+refusing them would refuse the defaults: ``EnableMultiThreadRowToBatch()`` on a read, a write's
+temporary directory and spill configuration, and ``WithIgnoreNumBucketCheck()`` and
+``WithIgnorePreviousFiles()`` on a write. They have no effect here: a format read hands out the
+batches parquet or orc already produced rather than assembling them from rows, a format write
+buffers in memory and never spills, and a table with no buckets has no bucket count to check and
+no previous files to read back.
+
+A write is two-phase, since a directory has no metadata to switch atomically: a file is written
+into a ``_temporary`` directory beside where it will end up, under a hidden name of its own, and
+only the commit renames it into place. That is the layout Java Paimon's
+``RenamingTwoPhaseOutputStream`` stages under. The directory and the name are both hidden, the
+convention a Hive-style directory uses for output that is not committed table data, and what a scan
+of this table skips. The ``_temporary`` directory is shared with every other writer
+of the same table and is left behind after a commit.
+
+A plan is in-memory only. ``FormatDataSplit`` has no serialized form - ``Split::Serialize()``
+refuses it - and neither has ``FormatCommitMessage``: a format table's plan has no cross-runtime
+encoding, so plan, read and commit within one process.
 
 A ``FormatTableWrite`` and a ``FormatTableCommit`` are each driven by one thread, but separate
 ones may fill and add to a table at once: each write stages its files under a uuid of its own, and
@@ -65,15 +133,20 @@ closer to the target.
 Aborting a write
 ----------------
 ``FormatTableWrite::Abort()`` removes the files the write staged. It is the one call still allowed
-after ``PrepareCommit()``, so a commit that is prepared and then abandoned can still be cleaned up.
+after ``PrepareCommit()``, and that is what it is for: a write dropped *before* preparing clears
+its staged files from its own destructor, so only a commit that is prepared and then abandoned
+needs it.
 
 Path containment is checked on the path text, which stops a ``..`` from leaving the table but not
 a symbolic link pointing out of it - the same as Java's own local file system behaviour.
 
 ``FormatTableCommit::Abort()`` does the same for the messages a commit was given. **Neither undoes
 a commit that succeeded**: once a file has been renamed into place it is no longer staged, and
-nothing here will take it back. Both are best effort and never fail, so a warning in the log is the
-only signal that a file could not be removed.
+nothing here will take it back. Java's committer removes the published path as well as the staged
+one, which matters there because a commit publishes file by file with nothing watching; here a
+commit that fails part way takes its own published files back before it returns, so an abort is
+left with the staged files alone. Both are best effort and never fail, so a warning in the log is
+the only signal that a file could not be removed.
 
 Give ``FormatTableCommit`` only the messages this job's own writers produced. A message names a
 staged file by path, and a commit can tell that the path belongs to this table, sits in the
@@ -102,14 +175,32 @@ instead. The one exception is the value standing for a null partition, ``partiti
 which the scan reads at a partition level by design. Under the ``key=value`` layout the question
 does not arise, since the key in front of the value keeps the directory name visible.
 
-Two smaller differences come from this library's own conventions:
+A few smaller differences come from this library's own conventions:
 
 * a write takes one partition per batch: the batch declares it through
   ``RecordBatch::SetPartition()``, every row is checked against that declaration, and a batch
   mixing partitions is refused. Java routes row by row, so one write call there may land in any
   number of partitions;
+* a write takes its partition from ``RecordBatch::SetPartition()`` rather than from the rows, so
+  the values arrive as text. They are still read into their column types and rendered back out
+  before anything is named after them - the round trip Java's writer makes when it renders a
+  partition out of the row it is writing, through the partition computer its
+  ``FileStorePathFactory`` holds. The table therefore decides the directory name and the commit
+  message, not the spelling the caller used: with ``partition.legacy-name`` on, its default, a
+  ``DATE`` partition is written as its day count whether the caller wrote ``19723`` or
+  ``2024-01-01``, and as ``YYYY-MM-DD`` when the option is off. A value that cannot be read into
+  its column type is refused. ``FormatTableCommit``'s static partition is *not* put through that
+  round trip and is used as given, which is what Java's ``FormatTableCommit.buildPartitionPath``
+  does with it too;
+* a commit message carries the partition its file belongs to, and a commit checks that it agrees
+  with the directory the file sits in. Java's message carries none and derives the partition from
+  the committer's target path, so the two cannot disagree there. A message here is a public struct
+  a caller may have built itself, so the value is checked rather than trusted;
 * a projection that names the same column twice is rejected when the read is built. Java reads
-  such a column once per entry.
+  such a column once per entry;
+* a row limit is the caller's: ``FormatTableScan`` takes one so that a plan can drop splits it
+  cannot need, but ``FormatTableRead`` does not bound the reader it hands out, and the caller stops
+  calling ``NextBatch()`` once it has enough. Java wraps its reader in a ``LimitRecordReader``.
 
 Current limits
 --------------
@@ -129,10 +220,13 @@ Compared with Java Paimon, this implementation does not yet support:
   Partition discovery here also lists one directory level at a time and applies the filter to each
   name, while Java turns a leading run of equality constraints into a path and starts listing
   below it; a table with many partitions therefore costs more listings here than in Java;
-* ``scan.ignore-corrupt-file`` and ``scan.ignore-lost-file``, which are not implemented: a
+* ``scan.ignore-corrupt-files`` and ``scan.ignore-lost-files``, which are not implemented: a
   corrupt or missing data file fails the read rather than being skipped;
-* ``partition.legacy-name``, which changes how a partition value is rendered into its directory
-  name;
+* ``dynamic-partition-overwrite``. An overwrite here always replaces the partitions the commit
+  actually writes to, which is what Java does under that option's default of ``true``. Java also
+  has the other mode: with it off, and always for an unpartitioned table, an overwrite empties
+  everything the table holds - so a statement whose query returns nothing still clears the table.
+  A commit with no messages therefore clears nothing here where Java would;
 * ``format-table.commit-hive-sync-url``, which registers committed partitions with a Hive
   metastore;
 * column default values. Java replaces a null in a column whose schema field declares a default
@@ -141,8 +235,12 @@ Compared with Java Paimon, this implementation does not yet support:
   out of what it writes, leaving files that carry nothing but a row count; here such a schema is
   refused when the table is created and when it is opened;
 * ``TIMESTAMP``, ``DECIMAL``, ``FLOAT`` and ``DOUBLE`` partition columns, which Java allows. This
-  is a restriction of the whole library rather than of format tables, and a table Java created
-  with such a partition column fails to open here rather than at the first read.
+  is a restriction of the whole library rather than of format tables. The types that do work are
+  ``BOOLEAN``, ``TINYINT``, ``SMALLINT``, ``INT``, ``BIGINT``, ``STRING`` and ``DATE`` - the set
+  the managed table path reads and writes partitions in. Any other, ``BINARY`` among them, is
+  refused when the table is created and when it is opened, rather than at the first read or write:
+  validation asks by building the partition computer that does the round trip, so there is one
+  answer rather than a list of types that could fall out of step with it.
 
 ``data-file.path-directory`` has no effect here, and none in Java either: Java's format table
 writer builds its paths from the table root rather than from that directory.

include/paimon/reader/batch_reader.h

@@ -47,10 +47,9 @@ class PAIMON_EXPORT BatchReader {
     /// avoid potential issues during conversion through the Arrow C Data Interface.
     ///
     /// @return A result containing a `::ReadBatch`, which consists of a unique pointer to
-    /// `ArrowArray` and a unique pointer to `ArrowSchema`. A reader over a table that records row
-    /// kinds puts a `_VALUE_KIND` field first to carry them; one over a table where every row is
-    /// an insert, such as a format table, has no such field. Deleted or index-filtered rows are
-    /// removed.
+    /// `ArrowArray` and a unique pointer to `ArrowSchema`. Returned array contains a `_VALUE_KIND`
+    /// field (the first field) to indicate the row kind of each row. Deleted or index-filtered rows
+    /// are removed.
     virtual Result<ReadBatch> NextBatch() = 0;
 
     /// Retrieves the next batch of data.
@@ -62,12 +61,12 @@ class PAIMON_EXPORT BatchReader {
     /// avoid potential issues during conversion through the Arrow C Data Interface.
     ///
     /// @return A result containing a `::ReadBatch` and a valid bitmap. `::ReadBatch` consists of a
-    /// unique pointer to `ArrowArray` and a unique pointer to `ArrowSchema`. The `_VALUE_KIND`
-    /// field is present or absent exactly as in `NextBatch()`. Deleted or index-filtered records
-    /// maybe maintained in `::ReadBatch`, while bitmap indicates valid row id. If deletion vector
-    /// or index are enabled, this function is more efficient than `NextBatch()`. The default
-    /// implementation calls `NextBatch()` and adds all rows to valid bitmap. Noted that the
-    /// returned bitmap has at least one valid row id.
+    /// unique pointer to `ArrowArray` and a unique pointer to `ArrowSchema`. Returned array
+    /// contains a _VALUE_KIND field (the first field) to indicate the row kind of each row. Deleted
+    /// or index-filtered records maybe maintained in `::ReadBatch`, while bitmap indicates valid
+    /// row id. If deletion vector or index are enabled, this function is more efficient than
+    /// `NextBatch()`. The default implementation calls `NextBatch()` and adds all rows to valid
+    /// bitmap. Noted that the returned bitmap has at least one valid row id.
     virtual Result<ReadBatchWithBitmap> NextBatchWithBitmap();
 
     /// Retrieves the reader's metrics.