feat: support real-time append writes with pluggable memory indexers

Author: lxy-9602

include/paimon/api.h

@@ -38,5 +38,10 @@
 #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/mem_indexer.h"
+#include "paimon/realtime/realtime_context.h"
+// IWYU pragma: end_exports
+
 /// Top-level namespace for Paimon C++ API.
 namespace paimon {}

include/paimon/file_store_commit.h

@@ -30,6 +30,7 @@
 #include "paimon/executor.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/metrics.h"
+#include "paimon/realtime/realtime_commit_progress.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
 #include "paimon/type_fwd.h"
@@ -71,6 +72,21 @@ class PAIMON_EXPORT FileStoreCommit {
                           int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER,
                           std::optional<int64_t> watermark = std::nullopt) = 0;
 
+    /// Commit sealed real-time segments and persist their partition-bucket offset progress.
+    ///
+    /// Entries for each partition-bucket must form a contiguous range beginning after the offset
+    /// recorded by the latest committed snapshot. Input entries may be unordered; this method
+    /// orders them by partition, bucket, and offset before validating continuity. The resulting
+    /// snapshot atomically publishes the data files and the updated offset map.
+    ///
+    /// @param realtime_commits Commit messages and inclusive offset ranges to commit.
+    /// @param commit_identifier Identifier of the streaming commit operation.
+    /// @param watermark Optional event-time watermark.
+    /// @return Status indicating the success or failure of the commit operation.
+    virtual Status CommitWithProgress(const std::vector<RealtimeCommitProgress>& realtime_commits,
+                                      int64_t commit_identifier,
+                                      std::optional<int64_t> watermark) = 0;
+
     /// Filter out all `std::vector<CommitMessage>` which have been committed and commit the
     /// remaining ones.
     ///

include/paimon/file_store_write.h

@@ -21,13 +21,15 @@
 #include <cstdint>
 #include <map>
 #include <memory>
+#include <string>
 #include <vector>
 
 #include "paimon/commit_message.h"
 #include "paimon/defs.h"
 #include "paimon/executor.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/metrics.h"
+#include "paimon/realtime/realtime_commit_progress.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
 #include "paimon/type_fwd.h"
@@ -85,8 +87,22 @@ class PAIMON_EXPORT FileStoreWrite {
     ///
     /// @return A Result containing `std::vector<std::shared_ptr<CommitMessage>>` objects,
     ///         representing the generated commit messages.
+    /// @note Real-time writers must use `PrepareCommitWithProgress()` so offset ranges are not
+    ///       discarded.
     virtual Result<std::vector<std::shared_ptr<CommitMessage>>> PrepareCommit(
         bool wait_compaction = true, int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER) = 0;
+
+    /// Generates commit messages together with partition-bucket real-time offset ranges.
+    ///
+    /// Each range is returned atomically with the commit message generated from the same sealed
+    /// segment.
+    ///
+    /// @param commit_identifier Identifier of this prepare-commit operation in streaming mode.
+    /// @return Real-time commit messages with their partition-bucket offset ranges.
+    /// @note Calling this method on a non-real-time writer or in batch mode returns an error.
+    virtual Result<std::vector<RealtimeCommitProgress>> PrepareCommitWithProgress(
+        int64_t commit_identifier);
+
     virtual std::shared_ptr<Metrics> GetMetrics() const = 0;
     virtual Status Close() = 0;
 };

include/paimon/realtime/arrow_mem_indexer_factory.h

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 "paimon/realtime/mem_indexer.h"
+
+namespace paimon {
+
+/// Factory for Paimon's default Arrow-backed `MemIndexer`.
+class PAIMON_EXPORT ArrowMemIndexerFactory : public MemIndexerFactory {
+ public:
+    /// Creates an Arrow-backed indexer for one partition and bucket.
+    Result<std::shared_ptr<MemIndexer>> Create(
+        ::ArrowSchema* write_schema, const std::map<std::string, std::string>& options,
+        const std::shared_ptr<MemoryPool>& memory_pool) override;
+};
+
+}  // namespace paimon

include/paimon/realtime/mem_indexer.h

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "paimon/reader/batch_reader.h"
+#include "paimon/record_batch.h"
+#include "paimon/result.h"
+#include "paimon/utils/range.h"
+#include "paimon/visibility.h"
+
+struct ArrowSchema;
+
+namespace paimon {
+
+class MemoryPool;
+
+/// A record batch and the contiguous offset range assigned to its rows.
+///
+/// The batch contains only the table write fields. `_OFFSET` is carried separately by
+/// `offset_range`; row `i` corresponds to `offset_range.from + i`. Paimon adds the physical
+/// `_OFFSET` column when the sealed segment is written to data files.
+struct PAIMON_EXPORT RealtimeWriteBatch {
+    /// Input batch whose ownership is transferred to `MemIndexer::Write`.
+    std::unique_ptr<RecordBatch> batch;
+    /// Inclusive `[from, to]` offset range covered by `batch`.
+    Range offset_range;
+};
+
+/// Opaque handle to an immutable segment returned by `MemIndexer::SealForCommit`.
+///
+/// A plugin may store the segment in memory or in spill files. Callers use this handle only to
+/// request commit readers and inspect its offset range.
+class PAIMON_EXPORT RealtimeSegmentHandle {
+ public:
+    virtual ~RealtimeSegmentHandle() = default;
+
+    /// Returns the inclusive offset range covered by this segment.
+    virtual Range GetOffsetRange() const = 0;
+};
+
+/// Plugin interface for buffering real-time writes before Paimon data-file generation.
+///
+/// Paimon serializes calls to `Write` and `SealForCommit` for the same indexer. After sealing,
+/// `CreateCommitReaders` may read the immutable sealed segment while later `Write` calls append to
+/// a new building segment. Paimon retains control of file format, rolling, indexes, and
+/// commit-message generation.
+class PAIMON_EXPORT MemIndexer {
+ public:
+    virtual ~MemIndexer() = default;
+
+    /// Adds a batch to the current building segment.
+    ///
+    /// The number of rows must equal the size of `offset_range`.
+    virtual Status Write(RealtimeWriteBatch&& batch) = 0;
+
+    /// Seals the current building data and opens a new building segment.
+    ///
+    /// Returns an immutable segment handle, or `std::nullopt` when there is no data to seal.
+    virtual Result<std::optional<std::shared_ptr<RealtimeSegmentHandle>>> SealForCommit() = 0;
+
+    /// Creates readers that expose all rows in a sealed segment for Paimon file writing.
+    ///
+    /// Concatenating the returned readers must produce every sealed row exactly once and in write
+    /// order. Each output batch contains `_VALUE_KIND` followed by all fields from the factory's
+    /// `write_schema`; it does not contain `_OFFSET`.
+    virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
+        const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;
+
+    /// Returns the number of bytes currently retained by the building segment.
+    virtual uint64_t GetMemoryUsage() const = 0;
+
+    /// Releases resources owned by this indexer and rejects subsequent writes or seals.
+    virtual Status Close() = 0;
+};
+
+/// Factory for application-provided `MemIndexer` implementations.
+class PAIMON_EXPORT MemIndexerFactory {
+ public:
+    virtual ~MemIndexerFactory() = default;
+
+    /// Creates an indexer configured with the supplied schema, options, and memory pool.
+    ///
+    /// `write_schema` uses the Arrow C Data Interface and contains the table fields accepted by
+    /// `Write`. It is valid only during this call. An implementation may consume its contents by
+    /// using an Arrow C Data Interface importer; otherwise Paimon releases them after this method
+    /// returns.
+    /// @param write_schema Table write schema without Paimon-generated real-time fields.
+    /// @param options Effective table options available to the indexer.
+    /// @param memory_pool Memory pool provided by the write context.
+    virtual Result<std::shared_ptr<MemIndexer>> Create(
+        ::ArrowSchema* write_schema, const std::map<std::string, std::string>& options,
+        const std::shared_ptr<MemoryPool>& memory_pool) = 0;
+};
+
+}  // namespace paimon

include/paimon/realtime/realtime_commit_progress.h

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <memory>
+#include <string>
+
+#include "paimon/commit_message.h"
+#include "paimon/utils/range.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A real-time commit message and its partition-bucket offset progress.
+///
+/// Offsets are scoped to one partition and bucket. `offset_range` is inclusive and covers all
+/// rows represented by `commit_message`. The progress fields are not embedded in
+/// `CommitMessage` serialization.
+struct PAIMON_EXPORT RealtimeCommitProgress {
+    /// Paimon commit message generated from one sealed segment.
+    std::shared_ptr<CommitMessage> commit_message;
+    /// Partition path such as `dt=2`, or an empty string for an unpartitioned table.
+    std::string partition;
+    /// Bucket containing the sealed segment.
+    int32_t bucket;
+    /// Inclusive offset range represented by the commit message.
+    Range offset_range;
+};
+
+}  // namespace paimon

include/paimon/realtime/realtime_context.h

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <memory>
+#include <string>
+
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+struct ArrowSchema;
+
+namespace paimon {
+
+class MemIndexer;
+class MemIndexerFactory;
+class MemoryPool;
+
+/// Shared context that owns the `MemIndexer` instances used by a real-time writer.
+///
+/// Applications attach one context to `WriteContext`. The context uses either the default Arrow
+/// implementation or an application-provided factory and keeps each created indexer available
+/// across multiple writes and prepare-commit operations.
+class PAIMON_EXPORT RealtimeContext {
+ public:
+    /// Creates a context backed by Paimon's default Arrow `MemIndexer`.
+    static Result<std::shared_ptr<RealtimeContext>> Create();
+
+    /// Creates a context backed by an application-provided indexer factory.
+    ///
+    /// @param factory Non-null factory used to create indexers on demand.
+    static Result<std::shared_ptr<RealtimeContext>> Create(
+        const std::shared_ptr<MemIndexerFactory>& factory);
+
+    /// Returns the stable indexer associated with a partition and bucket, creating it if needed.
+    Result<std::shared_ptr<MemIndexer>> GetOrCreateMemIndexer(
+        const std::string& partition, int32_t bucket, std::unique_ptr<::ArrowSchema> write_schema,
+        const std::map<std::string, std::string>& options,
+        const std::shared_ptr<MemoryPool>& memory_pool);
+
+    ~RealtimeContext();
+
+ private:
+    class Impl;
+
+    explicit RealtimeContext(std::unique_ptr<Impl>&& impl);
+
+    std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace paimon

include/paimon/type_fwd.h

@@ -56,6 +56,7 @@ class FileStatus;
 class BasicFileStatus;
 
 class RecordBatch;
+class RealtimeContext;
 
 class FormatWriter;
 

include/paimon/utils/special_field_ids.h

@@ -39,9 +39,11 @@ class SpecialFieldIds {
     static const int32_t ROW_KIND = std::numeric_limits<int32_t>::max() - 3;
     /// Special field ID reserved for row ID. Value: INT32_MAX - 5
     static const int32_t ROW_ID = std::numeric_limits<int32_t>::max() - 5;
-
     /// Special field ID reserved for index score. Value: CPP_FIELD_ID_END - 1
     static const int32_t INDEX_SCORE = CPP_FIELD_ID_END - 1;
+    // TODO(xinyu.lxy): Add _OFFSET to the Java codebase so compaction preserves its information.
+    /// Special field ID reserved for real-time offset. Value: CPP_FIELD_ID_END - 2
+    static const int32_t OFFSET = CPP_FIELD_ID_END - 2;
 };
 
 }  // namespace paimon