| .. 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. | |
| Write | |
| ===== | |
| Batch writing requires the compute engine to specify the target ``partition``. | |
| For fixed-bucket tables, the engine must also assign a valid bucket to every | |
| ``RecordBatch``. In unaware-bucket and postpone-bucket modes, the writer can | |
| resolve the bucket automatically when it is omitted. | |
| Paimon C++ uses Apache Arrow as the :ref:`in-memory columnar format<memory-format>` | |
| to more efficiently support writing to disk columnar formats such as ORC, | |
| Parquet, and Avro, thereby improving write throughput. | |
| .. note:: | |
| Currently supported table types: | |
| - Append table | |
| - Primary Key table | |
| Changelog is produced on the write path according to the | |
| ``changelog-producer`` table option (``input``, ``full-compaction`` or | |
| ``lookup``), and the changelog files are committed with their snapshot, | |
| except by the overwrite flow, which ignores them and logs a warning naming | |
| the files it ignored. See :doc:`compaction` for the ``full-compaction`` and | |
| ``lookup`` producers. | |
| Bucketing Modes | |
| --------------- | |
| - Append tables: | |
| * Support ``bucket = -1`` (unaware-bucket mode) | |
| * Support ``bucket > 0`` (fixed bucket mode) | |
| - PK tables: | |
| * Support ``bucket = -2`` (postpone bucket mode) | |
| * Support ``bucket > 0`` (fixed bucket mode) | |
| .. note:: | |
| PK tables do not support dynamic bucketing (``bucket = -1``). | |
| RecordBatch Construction | |
| ------------------------ | |
| - The compute engine must: | |
| - Assign the correct ``partition`` for each row. | |
| - In fixed-bucket mode, apply the Paimon-consistent bucketing function, set a | |
| bucket in ``[0, bucket)``, and group rows into Arrow ``RecordBatch`` objects | |
| per partition-bucket combination. | |
| - In unaware-bucket mode (append table with ``bucket = -1``), an omitted bucket | |
| is resolved to ``0``. | |
| - In postpone-bucket mode (primary-key table with ``bucket = -2``), an omitted | |
| bucket is resolved to ``-2``. | |
| - Recommended practices: | |
| - Use schema-aligned Arrow arrays with explicit validity bitmaps and offsets. | |
| - Prefer batch sizes tuned for I/O throughput (e.g., tens to hundreds of MB per flush, depending on filesystem and cluster configuration). | |
| - Maintain stable sort orders within a batch only if required by downstream merge or compaction logic; otherwise avoid unnecessary ordering costs. | |
| Writing BLOB Columns | |
| ~~~~~~~~~~~~~~~~~~~~ | |
| A ``BLOB`` column is a ``LargeBinary`` field carrying Paimon's BLOB field | |
| metadata, an ``ARRAY<BLOB>`` column is a top-level ``List`` field whose element | |
| field carries it, and a ``MAP<kt, BLOB>`` column is a top-level ``Map`` field | |
| whose values are ``LargeBinary``. Build the BLOB field with | |
| ``paimon::Blob::ArrowField`` and import it into Arrow; the element field of an | |
| ``ARRAY<BLOB>`` column must keep that metadata: | |
| .. code-block:: cpp | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<::ArrowSchema> c_element, | |
| paimon::Blob::ArrowField("element", /*nullable=*/true)); | |
| PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Field> element, | |
| arrow::ImportField(c_element.get())); | |
| std::shared_ptr<arrow::Schema> schema = arrow::schema( | |
| {arrow::field("id", arrow::int32()), arrow::field("frames", arrow::list(element)), | |
| arrow::field("views", arrow::map(arrow::utf8(), element))}); | |
| For a column stored in blob files, which includes every ``ARRAY<BLOB>`` and | |
| ``MAP<kt, BLOB>`` column, each value, element or map value holds either the raw | |
| bytes or a serialized ``paimon::BlobDescriptor`` produced by | |
| ``paimon::Blob::ToDescriptor``; the writer copies the referenced data into the | |
| blob file. A BLOB column listed in ``blob-descriptor-field`` or | |
| ``blob-view-field`` keeps the reference in the data file instead, so the | |
| referenced data must remain available. | |
| If the referenced data cannot be reached, the write fails unless a write-null | |
| option covers the failure: ``blob-write-null-on-missing-file`` covers a | |
| referenced file that does not exist, and ``blob-write-null-on-fetch-failure`` | |
| covers any other failure to resolve the descriptor or open the data, including | |
| a missing file when the former is disabled and an offset past the end of the | |
| file for a descriptor with a dynamic length (``-1``). A covered value is written | |
| as NULL; in an ``ARRAY<BLOB>`` or a ``MAP<kt, BLOB>`` only that element or map | |
| value becomes NULL, not the array or map. | |
| Any other failure fails the write, such as a failure to write the blob file or | |
| to close a referenced file. As in Paimon Java, this includes a file too short | |
| for the range of a descriptor with a known length, which is only detected while | |
| the data is copied. | |
| Except for the placeholder marker defined below, a ``MAP<kt, BLOB>`` value with | |
| a repeated or null key also fails the write, before any of its data is written. | |
| See :doc:`data_types` for the table requirements and restrictions. | |
| A data-evolution write can update columns of existing rows. The write does not | |
| locate the updated rows itself: before the commit, each file in its commit | |
| messages must be assigned the first row id of the rows it covers. A write that | |
| carries only columns stored in blob files is always such an update, and | |
| committing it without that id fails. A BLOB column listed in | |
| ``blob-descriptor-field`` or ``blob-view-field`` is stored in the data file | |
| instead. Paimon C++ does not assign that id and its public API cannot set it, | |
| so the commit messages must be updated outside Paimon C++, for example by | |
| Paimon Java after ``CommitMessage::Serialize``. The serialized payload does not | |
| carry its serialization version, so send ``CommitMessage::CurrentVersion()`` | |
| with it, and pass ``CommitMessage::Deserialize`` the version the returned | |
| payload was serialized with. | |
| In such an update, a row whose BLOB or ``ARRAY<BLOB>`` value stays unchanged is | |
| marked with the reserved bytes ``_PAIMON_BLOB_PLACEHOLDER``: as the value itself | |
| for a BLOB column, or as the only element of the array for an ``ARRAY<BLOB>`` | |
| column. Every write stores a value equal to the reserved bytes as such a marker, | |
| so that value is not supported. A row whose ``MAP<kt, BLOB>`` value stays | |
| unchanged is marked with a map of exactly two entries with equal keys and null | |
| values, which no other map can hold since its keys must be unique. As in Paimon | |
| Java, a marker works whatever other columns the update carries, and its row | |
| keeps its value from the older files when read. | |
| Append the two entries of a ``MAP<kt, BLOB>`` marker one by one, for example | |
| with ``arrow::MapBuilder``, since a JSON object or a dictionary merges equal | |
| keys into one entry: | |
| .. code-block:: cpp | |
| // map_builder builds the MAP<STRING, BLOB> column of the update. | |
| auto* keys = static_cast<arrow::StringBuilder*>(map_builder->key_builder()); | |
| auto* values = static_cast<arrow::LargeBinaryBuilder*>(map_builder->item_builder()); | |
| ARROW_RETURN_NOT_OK(map_builder->Append()); | |
| for (int32_t i = 0; i < 2; ++i) { | |
| ARROW_RETURN_NOT_OK(keys->Append("k")); | |
| ARROW_RETURN_NOT_OK(values->AppendNull()); | |
| } | |
| .. note:: | |
| The C++ writer differs from Paimon Java in these respects: | |
| - A placeholder is identified by the reserved bytes, which can collide with a | |
| user value, or by the two-entry map marker above; Java uses a dedicated | |
| placeholder object. | |
| - A ``MAP<kt, BLOB>`` key cannot be null, as Arrow map keys are not | |
| nullable, and cannot be TIME; Java allows one null key and TIME keys. | |
| - A missing file is detected with ``FileSystem::Exists``. Java detects a | |
| missing file for an ``ARRAY<BLOB>`` element or a ``MAP<kt, BLOB>`` value | |
| only from an HTTP 404, and does not write NULL for a 404 under | |
| ``blob-write-null-on-fetch-failure`` alone. | |
| - A descriptor with a dynamic length is copied up to the file length read | |
| when it is opened, so data appended to the file during the copy is left | |
| out, and a file truncated during the copy fails the write. Java reads it | |
| until the end of the file. Its offset past the end of the file also fails | |
| to open, whereas Java writes an empty value for it when the file system | |
| can seek past the end of a file, as the local one can. | |
| Prepare Commit | |
| ---------------- | |
| The compute engine is responsible for triggering the writer nodes' ``PrepareCommit``. | |
| Triggering conditions depend on the engine's business needs and can follow either: | |
| - Streaming mode: time-based or periodic triggers (e.g., every N seconds). | |
| - Batch mode: trigger after all data in the batch has been written. | |
| Once the compute engine collects ``CommitMessages`` from all writer nodes, it | |
| can issue a ``Commit`` request to the control plane (management path) to create | |
| a new ``Snapshot``. | |
| Compatibility Goals | |
| ~~~~~~~~~~~~~~~~~~~ | |
| To ensure interoperability, the ``PrepareCommit`` result produced by Paimon C++ | |
| must be consumable by Paimon Java. Therefore: | |
| - The structure and semantics of ``CommitMessage`` must remain consistent with | |
| Java Paimon. | |
| - Any evolution of the Java-side ``CommitMessage`` schema must be tracked and | |
| validated on the C++ side to maintain cross-language compatibility. | |
| Interface Design in Paimon C++ | |
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| Unlike Java Paimon, Paimon C++ does not expose ``BinaryRow``-like types in its | |
| public interfaces. To preserve compatibility without leaking internal row | |
| representations, Paimon C++ provides ``CommitMessage`` only through: | |
| - Serialization: convert the internal commit state into a well-defined binary | |
| representation that matches Java Paimon's expectations. | |
| - Deserialization: parse the Java-compatible binary representation back into | |
| C++ commit structures for validation, replay, or tooling needs. | |
| This design ensures that: | |
| - Public APIs are independent of Java-specific row abstractions. | |
| - Cross-language commit payloads remain stable and versionable. | |
| - Internal data layouts can evolve without breaking external consumers. | |
| CommitMessage Contract | |
| ~~~~~~~~~~~~~~~~~~~~~~ | |
| The ``CommitMessage`` must encode all information required by the coordinator to | |
| produce a correct ``Snapshot``, which commonly includes (but is not limited to): | |
| - Partition and bucket identifiers associated with written data. | |
| - New data files, delete files (as applicable to the table type). | |
| - File-level metadata required for manifest and index updates (e.g., row counts, min/max statistics where applicable). | |
| - Transactional markers and sequence numbers as required by table semantics. | |
| - Any per-writer state necessary for deduplication or idempotent commits. | |
| .. note:: | |
| The C++ writer supports Append and PK tables and can produce | |
| ``CommitMessage`` objects for both. ``FileStoreCommit`` supports direct | |
| file-system commits for both table types on non-object-store paths. | |
| Object-store paths need a commit that does not rename: either a catalog that | |
| manages the versions of its tables, such as the REST catalog, passed to | |
| ``CommitContextBuilder::WithCatalog()``, or ``UseRESTCatalogCommit(true)``, | |
| which builds the commit request for the caller to send instead of publishing | |
| the snapshot itself. The two are mutually exclusive. A catalog without | |
| version management still commits by renaming, so passing one is rejected on | |
| an object-store path. Changelog files are produced on the write path and | |
| committed with their snapshot, except by the overwrite flow, which ignores | |
| them and logs a warning naming the files it ignored. | |
| Serialization and Deserialization | |
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| - **Binary format:** The binary payload must strictly conform to Java Paimon's | |
| ``CommitMessage`` encoding. It does not contain a version tag, so callers | |
| must transport ``CommitMessage::CurrentVersion()`` separately and supply it | |
| when deserializing. | |
| - **Serialization API:** Use ``CommitMessage::Serialize`` for one message or | |
| ``CommitMessage::SerializeList`` for a list. | |
| - **Deserialization API:** Use ``CommitMessage::Deserialize`` or | |
| ``CommitMessage::DeserializeList`` with the separately supplied | |
| serialization version. | |
| - **Validation:** Conformance and round-trip tests must verify compatibility | |
| with Java Paimon for supported message versions. | |
| Operational Flow | |
| ~~~~~~~~~~~~~~~~~~~~~~~ | |
| 1. Writer nodes perform data ingestion and produce Arrow ``RecordBatch`` | |
| organized by partition and bucket. | |
| 2. Writers flush batches into data files through the registered | |
| ``file.format`` and ``file-system`` backends, producing file-level metadata | |
| and per-batch commit state. | |
| 3. Each writer invokes ``PrepareCommit``, which: | |
| - Aggregates per-writer state into ``CommitMessage`` objects. | |
| - Returns ``CommitMessage`` objects; it does not serialize them. | |
| 4. The compute engine gathers ``CommitMessage`` objects from all writers. For | |
| cross-process transport, it explicitly calls ``Serialize`` or | |
| ``SerializeList`` and carries ``CurrentVersion()`` alongside the payload. | |
| 5. For a direct file-system commit on a non-object-store path, the engine | |
| passes the objects to ``FileStoreCommit`` for either an Append or PK table. | |
| For a table of a catalog which owns its versions, such as a REST catalog, it | |
| passes that catalog and the table identifier to | |
| ``WriteContextBuilder::WithCatalog`` and ``CommitContextBuilder::WithCatalog``. | |
| The writer loads the latest snapshot from the catalog, and the current schema | |
| from it as well when it writes to the main branch; a write aimed at another | |
| branch reads the schema published under ``branch/branch-<name>`` instead. On | |
| ``Commit``, the snapshot and the statistics of the change go to the catalog, | |
| which decides whether this commit wins, and one that lost the race is rebased | |
| and retried the way a file-system commit is. A caller without a catalog | |
| client can instead enable REST catalog commit mode, call ``Commit``, obtain | |
| the JSON request from ``GetLastCommitTableRequest``, and send that request to | |
| the REST catalog itself. That request body names no branch, so a commit aimed | |
| at one has to be sent to the URL of ``tbl$branch_dev`` rather than of ``tbl``. | |
| 6. The local committer or REST catalog validates the messages, updates | |
| manifests/metadata, and finalizes the snapshot atomically. |