feat(realtime): add primary-key in-memory support

include/paimon/realtime/arrow_realtime_store_factory.h

@@ -26,11 +26,7 @@ namespace paimon {
 /// Factory for Paimon's default Arrow-backed `RealtimeStore`.
 class PAIMON_EXPORT ArrowRealtimeStoreFactory : public RealtimeStoreFactory {
  public:
-    /// Creates an Arrow-backed store for one partition and bucket.
-    Result<std::shared_ptr<RealtimeStore>> Create(
-        std::unique_ptr<::ArrowSchema> write_schema, StatisticsMode statistics_mode,
-        const std::map<std::string, std::string>& options,
-        const std::shared_ptr<MemoryPool>& memory_pool) override;
+    Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) override;
 };
 
 }  // namespace paimon

include/paimon/realtime/realtime_store.h

@@ -25,8 +25,10 @@
 #include <optional>
 #include <string>
 #include <utility>
+#include <variant>
 #include <vector>
 
+#include "arrow/c/abi.h"
 #include "paimon/reader/batch_reader.h"
 #include "paimon/realtime/offset_range.h"
 #include "paimon/record_batch.h"
@@ -41,10 +43,36 @@ namespace paimon {
 class MemoryPool;
 class Predicate;
 
-/// A table record batch and its framework-assigned contiguous offset range.
+struct PAIMON_EXPORT AppendRealtimeStoreCreateConfig {
+    StatisticsMode statistics_mode;
+};
+
+struct PAIMON_EXPORT PrimaryKeyRealtimeStoreCreateConfig {
+    /// Primary-key fields after removing partition fields, in comparison order.
+    std::vector<std::string> trimmed_primary_keys;
+};
+
+using RealtimeStoreCreateConfig =
+    std::variant<AppendRealtimeStoreCreateConfig, PrimaryKeyRealtimeStoreCreateConfig>;
+
+struct PAIMON_EXPORT RealtimeStoreCreateRequest {
+    /// Schema whose ownership is transferred to the factory. Append mode receives the complete
+    /// table write schema. Primary-key mode receives the prepared transport schema:
+    /// [_VALUE_KIND, _SEQUENCE_NUMBER, _REALTIME_OFFSET, table write fields].
+    std::unique_ptr<::ArrowSchema> write_schema;
+    std::map<std::string, std::string> options;
+    std::shared_ptr<MemoryPool> memory_pool;
+    std::map<std::string, std::string> partition;
+    int32_t bucket = -1;
+    RealtimeStoreCreateConfig mode_config;
+};
+
+/// A record batch and its framework-assigned contiguous offset range.
 ///
-/// The batch contains only table write fields. Row `i` is associated with
-/// `offset_range.begin + i`; the offset is progress metadata and is not a table field.
+/// Append-mode batches contain table write fields, and row `i` has offset
+/// `offset_range.begin + i`. Primary-key batches use the prepared transport schema, are sorted
+/// by full primary key then sequence number, and retain the original offset in
+/// `_REALTIME_OFFSET`.
 struct PAIMON_EXPORT RealtimeWriteBatch {
     /// Input batch whose ownership is transferred to `RealtimeStore::Write`.
     std::unique_ptr<RecordBatch> batch;
@@ -79,7 +107,10 @@ class PAIMON_EXPORT RealtimeReadView {
 
 /// Parameters used by a `RealtimeStore` to create readers for a query.
 struct PAIMON_EXPORT RealtimeQueryContext {
-    /// Requested output fields before the mandatory leading `_VALUE_KIND` field is added.
+    /// Append mode receives the requested output fields before the mandatory leading
+    /// `_VALUE_KIND` field is added. Primary-key mode receives the complete prepared schema.
+    /// This schema is borrowed and remains valid only during `CreateQueryReaders`; plugins must
+    /// import or copy it synchronously.
     ::ArrowSchema* read_schema;
     /// Predicate using field indexes from `read_schema`.
     std::shared_ptr<Predicate> predicate;
@@ -116,9 +147,10 @@ class PAIMON_EXPORT RealtimeStore {
 
     /// 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`.
+    /// The returned readers collectively expose every sealed row exactly once. Append-mode readers
+    /// preserve write order and contain `_VALUE_KIND` followed by table write fields. Primary-key
+    /// readers use the prepared transport schema; each reader's complete stream is sorted by full
+    /// primary key then sequence number.
     virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
         const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;
 
@@ -128,13 +160,15 @@ class PAIMON_EXPORT RealtimeStore {
     /// also provide a consistent snapshot when a write or seal is in progress.
     virtual Result<std::shared_ptr<RealtimeReadView>> AcquireReadView() = 0;
 
-    /// Creates readers over rows in `view` whose offsets are greater than or equal to
-    /// `offset_begin`.
+    /// Creates readers over rows in `view`. Append mode returns rows whose offsets are greater than
+    /// or equal to `offset_begin`; primary-key mode ignores `offset_begin`.
     ///
-    /// Each output batch contains `_VALUE_KIND` first, followed by the fields requested by
-    /// `context.read_schema` except a duplicate `_VALUE_KIND`. Concatenating all returned readers
-    /// must produce every matching row once. Paimon retains `view` for the lifetime of the
-    /// resulting framework reader.
+    /// Append-mode batches contain `_VALUE_KIND` followed by the requested fields except a duplicate
+    /// `_VALUE_KIND`, and collectively expose every matching row exactly once. Primary-key batches
+    /// use the prepared transport schema and may contain multiple mutations per key; each reader's
+    /// complete stream is sorted by full primary key then sequence number, and the readers
+    /// collectively expose every raw mutation exactly once. Paimon retains `view` for the lifetime
+    /// of the resulting framework reader.
     virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateQueryReaders(
         const std::shared_ptr<RealtimeReadView>& view, int64_t offset_begin,
         const RealtimeQueryContext& context) = 0;
@@ -157,16 +191,9 @@ class PAIMON_EXPORT RealtimeStoreFactory {
  public:
     virtual ~RealtimeStoreFactory() = default;
 
-    /// Creates a store configured with the supplied schema, statistics, options, and memory pool.
-    /// @param write_schema Complete table write schema whose ownership is transferred to the
-    /// factory. The factory may consume it or retain it in the created store.
-    /// @param statistics_mode Framework-parsed statistics collection mode.
-    /// @param options Effective table options available to the store.
-    /// @param memory_pool Memory pool provided by the write context.
-    virtual Result<std::shared_ptr<RealtimeStore>> Create(
-        std::unique_ptr<::ArrowSchema> write_schema, StatisticsMode statistics_mode,
-        const std::map<std::string, std::string>& options,
-        const std::shared_ptr<MemoryPool>& memory_pool) = 0;
+    /// Creates a store for the requested table mode and partition-bucket.
+    /// The factory consumes `request`, including ownership of `request.write_schema`.
+    virtual Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) = 0;
 };
 
 }  // namespace paimon

include/paimon/utils/special_field_ids.h

@@ -42,6 +42,8 @@ class SpecialFieldIds {
 
     /// Special field ID reserved for index score. Value: CPP_FIELD_ID_END - 1
     inline static constexpr int32_t INDEX_SCORE = CPP_FIELD_ID_END - 1;
+    /// Special field ID reserved for realtime offset. Value: CPP_FIELD_ID_END - 2
+    inline static constexpr int32_t REALTIME_OFFSET = CPP_FIELD_ID_END - 2;
 
     /// Lowest field ID reserved for system fields; IDs at or above it are excluded from the
     /// highest field ID of a schema. Value: INT32_MAX / 2

src/paimon/CMakeLists.txt

@@ -378,9 +378,12 @@ set(PAIMON_CORE_SRCS
     core/operation/write_restore.cpp
     core/realtime/arrow_realtime_store.cpp
     core/realtime/arrow_realtime_store_factory.cpp
+    core/realtime/prepared_key_value_reader.cpp
+    core/realtime/primary_key_realtime_store.cpp
     core/realtime/realtime_append_only_writer.cpp
     core/realtime/realtime_context.cpp
     core/realtime/realtime_context_impl.cpp
+    core/realtime/realtime_primary_key_writer.cpp
     core/postpone/postpone_bucket_writer.cpp
     core/schema/arrow_schema_validator.cpp
     core/schema/schema_manager.cpp
@@ -780,6 +783,7 @@ if(PAIMON_BUILD_TESTS)
                     core/manifest/index_manifest_file_handler_test.cpp
                     core/memory/writer_memory_manager_test.cpp
                     core/realtime/arrow_realtime_store_test.cpp
+                    core/realtime/primary_key_realtime_store_test.cpp
                     core/realtime/realtime_context_test.cpp
                     core/realtime/realtime_reader_test.cpp
                     core/mergetree/levels_test.cpp

src/paimon/core/io/merged_key_value_record_reader.cpp

@@ -117,13 +117,21 @@ Status MergedKeyValueRecordReader::Iterator::LoadNextRawKeyValue() const {
 }
 
 Result<std::unique_ptr<KeyValueRecordReader::Iterator>> MergedKeyValueRecordReader::NextBatch() {
+    if (initialization_error_.has_value()) {
+        return initialization_error_.value();
+    }
     if (visited_) {
         return std::unique_ptr<KeyValueRecordReader::Iterator>();
     }
     visited_ = true;
 
     auto iterator = std::make_unique<Iterator>(this);
-    PAIMON_ASSIGN_OR_RAISE(bool has_next, iterator->HasNext());
+    Result<bool> has_next_result = iterator->HasNext();
+    if (!has_next_result.ok()) {
+        initialization_error_ = has_next_result.status();
+        return initialization_error_.value();
+    }
+    bool has_next = std::move(has_next_result).value();
     if (!has_next) {
         return std::unique_ptr<KeyValueRecordReader::Iterator>();
     }

src/paimon/core/io/merged_key_value_record_reader.h

@@ -67,6 +67,7 @@ class MergedKeyValueRecordReader : public KeyValueRecordReader {
 
  private:
     bool visited_ = false;
+    std::optional<Status> initialization_error_;
     std::unique_ptr<KeyValueRecordReader> reader_;
     std::shared_ptr<FieldsComparator> key_comparator_;
     std::shared_ptr<MergeFunctionWrapper<KeyValue>> merge_function_wrapper_;