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

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,28 @@ 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);
+
+    /// Refreshes a real-time writer after `snapshot_id` has committed successfully.
+    ///
+    /// The writer loads the snapshot's partition-bucket offsets and releases sealed memory that is
+    /// fully covered by disk. Calling this method on a non-real-time writer returns an error.
+    virtual Status RefreshCommittedSnapshot(int64_t snapshot_id);
+
     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,165 @@
+/*
+ * 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;
+class Predicate;
+
+/// 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;
+};
+
+/// Opaque immutable view of the rows visible from one `MemIndexer`.
+///
+/// A view pins all referenced resources until the readers created from it are closed. Later
+/// writes, seals, and committed-offset reclamation do not change the contents of an existing view.
+class PAIMON_EXPORT MemReadView {
+ public:
+    virtual ~MemReadView() = default;
+
+    /// Returns the inclusive offset range visible in this view, or no range when it is empty.
+    virtual std::optional<Range> GetOffsetRange() const = 0;
+};
+
+/// Parameters used by a `MemIndexer` to create readers for a query.
+struct PAIMON_EXPORT MemQueryContext {
+    /// Requested output fields before the mandatory leading `_VALUE_KIND` field is added.
+    ///
+    /// The schema uses the Arrow C Data Interface and is valid only during
+    /// `CreateQueryReaders`. An implementation may consume it with an Arrow importer.
+    ::ArrowSchema* read_schema;
+    /// Predicate using field indexes from `read_schema`.
+    std::shared_ptr<Predicate> predicate;
+    /// Whether exact predicate filtering is enabled for the returned readers.
+    ///
+    /// Keep this disabled for primary-key merge-on-read. Filtering memory before PK merge may
+    /// remove the newest row and incorrectly expose an older disk row.
+    bool enable_predicate_filter;
+};
+
+/// 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;
+
+    /// Acquires an immutable view containing the current sealed and building rows.
+    ///
+    /// This method may be called concurrently with query-reader creation and reclamation. It must
+    /// also provide a consistent snapshot when a write or seal is in progress.
+    virtual Result<std::shared_ptr<MemReadView>> AcquireReadView() = 0;
+
+    /// Creates readers over rows in `view` whose offsets are greater than
+    /// `offset_lower_exclusive`.
+    ///
+    /// Each output batch contains `_VALUE_KIND` first, followed by the fields requested by
+    /// `context.read_schema` except a duplicate `_VALUE_KIND`. `_OFFSET` is returned when requested
+    /// by the read schema. Concatenating all returned readers must produce every matching row once.
+    virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateQueryReaders(
+        const std::shared_ptr<MemReadView>& view, int64_t offset_lower_exclusive,
+        const MemQueryContext& context) = 0;
+
+    /// Releases this indexer's ownership of sealed segments fully covered by the committed offset.
+    /// Existing read views continue to keep their referenced resources alive.
+    virtual Status Reclaim(int64_t committed_offset) = 0;
+
+    /// Returns the number of bytes currently retained by building and sealed segments.
+    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,46 @@
+/*
+ * 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/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;
+    /// Logical partition values, or an empty map for an unpartitioned table.
+    std::map<std::string, std::string> partition;
+    /// Bucket containing the sealed segment.
+    int32_t bucket;
+    /// Inclusive offset range represented by the commit message.
+    Range offset_range;
+};
+
+}  // namespace paimon