diff --git a/include/paimon/realtime/arrow_realtime_store_factory.h b/include/paimon/realtime/arrow_realtime_store_factory.h
index 4d65743ab..b3fa630de 100644
--- a/include/paimon/realtime/arrow_realtime_store_factory.h
+++ b/include/paimon/realtime/arrow_realtime_store_factory.h
@@ -27,10 +27,7 @@ namespace paimon {
 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
diff --git a/include/paimon/realtime/realtime_store.h b/include/paimon/realtime/realtime_store.h
index d02952acd..6ae81f1f4 100644
--- a/include/paimon/realtime/realtime_store.h
+++ b/include/paimon/realtime/realtime_store.h
@@ -27,6 +27,7 @@
 #include <utility>
 #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 +42,33 @@ namespace paimon {
 class MemoryPool;
 class Predicate;
 
-/// A table record batch and its framework-assigned contiguous offset range.
+enum class PAIMON_EXPORT RealtimeStoreMode {
+    APPEND_ONLY,
+    PRIMARY_KEY,
+};
+
+/// Parameters used by a `RealtimeStoreFactory` to create a store.
+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 realtime primary-key transport schema:
+    /// [_VALUE_KIND, _SEQUENCE_NUMBER, _REALTIME_OFFSET, table write fields].
+    std::unique_ptr<::ArrowSchema> write_schema;
+    /// Table options available to the store implementation.
+    std::map<std::string, std::string> options;
+    /// Memory pool for allocations retained by the store.
+    std::shared_ptr<MemoryPool> memory_pool;
+    /// Table mode implemented by the store.
+    RealtimeStoreMode mode = RealtimeStoreMode::APPEND_ONLY;
+    /// Statistics collected by append-only stores.
+    StatisticsMode statistics_mode = StatisticsMode::NONE;
+};
+
+/// 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 realtime primary-key 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 +103,11 @@ 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 requested realtime primary-key
+    /// transport 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 +144,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 realtime primary-key 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 +157,16 @@ 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 requested realtime primary-key transport schema, including
+    /// nested field-ID alignment, 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;
@@ -158,15 +190,9 @@ class PAIMON_EXPORT RealtimeStoreFactory {
     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.
+    /// The factory consumes `request`, including ownership of `request.write_schema`.
+    virtual Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) = 0;
 };
 
 }  // namespace paimon
diff --git a/include/paimon/utils/special_field_ids.h b/include/paimon/utils/special_field_ids.h
index 829f29889..5219d72db 100644
--- a/include/paimon/utils/special_field_ids.h
+++ b/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
diff --git a/src/paimon/common/table/special_fields.h b/src/paimon/common/table/special_fields.h
index 74b95b19c..8908f9f0c 100644
--- a/src/paimon/common/table/special_fields.h
+++ b/src/paimon/common/table/special_fields.h
@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <limits>
 #include <string>
+#include <utility>
 
 #include "arrow/type_fwd.h"
 #include "paimon/common/types/data_field.h"
@@ -66,13 +67,20 @@ struct SpecialFields {
         return data_field;
     }
 
+    static const DataField& RealtimeOffset() {
+        static const DataField data_field =
+            DataField(SpecialFieldIds::REALTIME_OFFSET,
+                      arrow::field("_REALTIME_OFFSET", arrow::int64(), false));
+        return data_field;
+    }
+
     static bool IsSystemField(const std::string& field_name) {
         if (StringUtils::StartsWith(field_name, KEY_FIELD_PREFIX)) {
             return true;
         }
         return field_name == SequenceNumber().Name() || field_name == ValueKind().Name() ||
                field_name == RowKind().Name() || field_name == RowId().Name() ||
-               field_name == IndexScore().Name();
+               field_name == IndexScore().Name() || field_name == RealtimeOffset().Name();
     }
 
     // TODO(xinyu.lxy): add a func to complete row-tracking fields
diff --git a/src/paimon/common/table/special_fields_test.cpp b/src/paimon/common/table/special_fields_test.cpp
index 68e805fd6..58a025ba2 100644
--- a/src/paimon/common/table/special_fields_test.cpp
+++ b/src/paimon/common/table/special_fields_test.cpp
@@ -55,6 +55,13 @@ TEST(SpecialFieldsTest, TestIndexScore) {
     ASSERT_EQ(SpecialFields::IndexScore().Type()->id(), arrow::Type::FLOAT);
 }
 
+TEST(SpecialFieldsTest, TestRealtimeOffset) {
+    ASSERT_EQ(SpecialFields::RealtimeOffset().Id(), SpecialFieldIds::REALTIME_OFFSET);
+    ASSERT_EQ(SpecialFields::RealtimeOffset().Name(), "_REALTIME_OFFSET");
+    ASSERT_EQ(SpecialFields::RealtimeOffset().Type()->id(), arrow::Type::INT64);
+    ASSERT_FALSE(SpecialFields::RealtimeOffset().Nullable());
+}
+
 TEST(SpecialFieldsTest, TestKeyValueSpecialFieldCount) {
     ASSERT_EQ(SpecialFields::KEY_VALUE_SPECIAL_FIELD_COUNT, 2);
 }
@@ -66,6 +73,7 @@ TEST(SpecialFieldsTest, TestIsSystemField) {
     ASSERT_TRUE(SpecialFields::IsSystemField("rowkind"));
     ASSERT_TRUE(SpecialFields::IsSystemField("_ROW_ID"));
     ASSERT_TRUE(SpecialFields::IsSystemField("_INDEX_SCORE"));
+    ASSERT_TRUE(SpecialFields::IsSystemField("_REALTIME_OFFSET"));
     ASSERT_TRUE(SpecialFields::IsSystemField("_KEY_0"));
 }
 
diff --git a/src/paimon/core/realtime/arrow_realtime_store_factory.cpp b/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
index 1d7219c41..edd8cfeee 100644
--- a/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
+++ b/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
@@ -30,20 +30,29 @@
 namespace paimon {
 
 Result<std::shared_ptr<RealtimeStore>> ArrowRealtimeStoreFactory::Create(
-    std::unique_ptr<ArrowSchema> write_schema, StatisticsMode statistics_mode,
-    const std::map<std::string, std::string>&, const std::shared_ptr<MemoryPool>& memory_pool) {
-    if (!write_schema || !write_schema->release) {
+    RealtimeStoreCreateRequest&& request) {
+    if (!request.write_schema || !request.write_schema->release) {
         return Status::Invalid("real-time store write schema is null");
     }
-    ScopeGuard schema_guard([schema = write_schema.get()]() { ArrowSchemaRelease(schema); });
-    if (!memory_pool) {
+    ScopeGuard schema_guard(
+        [schema = request.write_schema.get()]() { ArrowSchemaRelease(schema); });
+    if (!request.memory_pool) {
         return Status::Invalid("real-time store memory pool is null");
     }
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> imported_schema,
-                                      arrow::ImportSchema(write_schema.get()));
-    std::shared_ptr<arrow::MemoryPool> arrow_pool = GetArrowPool(memory_pool);
-    return std::make_shared<ArrowRealtimeStore>(imported_schema, statistics_mode, memory_pool,
-                                                arrow_pool);
+                                      arrow::ImportSchema(request.write_schema.get()));
+    switch (request.mode) {
+        case RealtimeStoreMode::APPEND_ONLY: {
+            std::shared_ptr<arrow::MemoryPool> arrow_pool = GetArrowPool(request.memory_pool);
+            return std::make_shared<ArrowRealtimeStore>(imported_schema, request.statistics_mode,
+                                                        request.memory_pool, arrow_pool);
+        }
+        case RealtimeStoreMode::PRIMARY_KEY: {
+            return Status::NotImplemented(
+                "primary-key real-time store support is not installed");
+        }
+    }
+    return Status::Invalid("invalid real-time store mode: ", static_cast<int32_t>(request.mode));
 }
 
 }  // namespace paimon
diff --git a/src/paimon/core/realtime/arrow_realtime_store_test.cpp b/src/paimon/core/realtime/arrow_realtime_store_test.cpp
index 9aae99332..864b1f810 100644
--- a/src/paimon/core/realtime/arrow_realtime_store_test.cpp
+++ b/src/paimon/core/realtime/arrow_realtime_store_test.cpp
@@ -232,8 +232,11 @@ TEST_F(ArrowRealtimeStoreTest, TestCommitReaderPreservesSlicedBatch) {
 TEST_F(ArrowRealtimeStoreTest, TestFullStatisticsPrunesNonMatchingBatch) {
     ArrowRealtimeStoreFactory factory;
     std::unique_ptr<ArrowSchema> write_schema = MakeReadSchema(schema_);
+    RealtimeStoreCreateRequest request{std::move(write_schema),
+                                       /*options=*/{}, pool_, RealtimeStoreMode::APPEND_ONLY,
+                                       StatisticsMode::FULL};
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeStore> realtime_store,
-                         factory.Create(std::move(write_schema), StatisticsMode::FULL, {}, pool_));
+                         factory.Create(std::move(request)));
     std::shared_ptr<ArrowRealtimeStore> store =
         std::dynamic_pointer_cast<ArrowRealtimeStore>(realtime_store);
     ASSERT_NE(nullptr, store);
diff --git a/src/paimon/core/realtime/realtime_append_only_writer.cpp b/src/paimon/core/realtime/realtime_append_only_writer.cpp
index 9d519d791..632d64e16 100644
--- a/src/paimon/core/realtime/realtime_append_only_writer.cpp
+++ b/src/paimon/core/realtime/realtime_append_only_writer.cpp
@@ -55,10 +55,11 @@ Result<std::shared_ptr<RealtimeAppendOnlyWriter>> RealtimeAppendOnlyWriter::Crea
     }
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeContextImpl> realtime_context_impl,
                            RealtimeContextImpl::Cast(realtime_context));
-    PAIMON_ASSIGN_OR_RAISE(
-        RealtimeStoreState store_state,
-        realtime_context_impl->GetOrCreateRealtimeStore(partition, bucket, std::move(write_schema),
-                                                        statistics_mode, options, memory_pool));
+    RealtimeStoreCreateRequest request{std::move(write_schema), options, memory_pool,
+                                       RealtimeStoreMode::APPEND_ONLY, statistics_mode};
+    PAIMON_ASSIGN_OR_RAISE(RealtimeStoreState store_state,
+                           realtime_context_impl->GetOrCreateRealtimeStore(
+                               std::move(request), RealtimePartitionBucket(partition, bucket)));
     return std::shared_ptr<RealtimeAppendOnlyWriter>(new RealtimeAppendOnlyWriter(
         store_state.store, file_writer, input_schema, store_state.initial_offset, memory_pool));
 }
diff --git a/src/paimon/core/realtime/realtime_context_impl.cpp b/src/paimon/core/realtime/realtime_context_impl.cpp
index f6bad5cf1..2b9d7d07c 100644
--- a/src/paimon/core/realtime/realtime_context_impl.cpp
+++ b/src/paimon/core/realtime/realtime_context_impl.cpp
@@ -34,14 +34,33 @@
 #include <utility>
 #include <vector>
 
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
 #include "arrow/c/helpers.h"
+#include "fmt/format.h"
 #include "paimon/arrow/abi.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
 #include "paimon/common/utils/uuid.h"
 #include "paimon/macros.h"
 #include "paimon/realtime/realtime_store.h"
 #include "paimon/status.h"
 
 namespace paimon {
+namespace {
+
+std::string PartitionToString(const std::map<std::string, std::string>& partition) {
+    std::string result = "{";
+    for (auto iter = partition.begin(); iter != partition.end(); ++iter) {
+        if (iter != partition.begin()) {
+            result += ", ";
+        }
+        result += iter->first + "=" + iter->second;
+    }
+    return result + "}";
+}
+
+}  // namespace
 
 RealtimeContextImpl::RealtimeContextImpl(const std::shared_ptr<RealtimeStoreFactory>& factory)
     : factory_(factory) {}
@@ -78,31 +97,36 @@ Status RealtimeContextImpl::Start() {
 }
 
 Result<RealtimeStoreState> RealtimeContextImpl::GetOrCreateRealtimeStore(
-    const std::map<std::string, std::string>& partition, int32_t bucket,
-    std::unique_ptr<ArrowSchema> write_schema, StatisticsMode statistics_mode,
-    const std::map<std::string, std::string>& options,
-    const std::shared_ptr<MemoryPool>& memory_pool) {
+    RealtimeStoreCreateRequest&& request, const RealtimePartitionBucket& partition_bucket) {
+    if (!request.write_schema || !request.write_schema->release) {
+        return Status::Invalid("real-time store write schema is null");
+    }
+    ScopeGuard schema_guard(
+        [schema = request.write_schema.get()]() { ArrowSchemaRelease(schema); });
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> requested_schema,
+                                      arrow::ImportSchema(request.write_schema.get()));
+    schema_guard.Release();
     std::lock_guard<std::mutex> progress_lock(progress_mutex_);
     std::lock_guard<std::mutex> registry_lock(mutex_);
-    const RealtimePartitionBucket key(partition, bucket);
+    auto iter = stores_.find(partition_bucket);
     int64_t initial_offset = 0;
-    auto offset_iter = committed_offsets_.find(key);
+    auto offset_iter = committed_offsets_.find(partition_bucket);
     if (offset_iter != committed_offsets_.end()) {
         if (offset_iter->second == std::numeric_limits<int64_t>::max()) {
-            if (write_schema) {
-                ArrowSchemaRelease(write_schema.get());
-            }
             return Status::Invalid("real-time offset has reached INT64_MAX");
         }
         initial_offset = offset_iter->second;
     }
-    auto iter = stores_.find(key);
     if (iter != stores_.end()) {
-        if (write_schema) {
-            ArrowSchemaRelease(write_schema.get());
+        if (iter->second.mode != request.mode ||
+            !iter->second.write_schema->Equals(*requested_schema, /*check_metadata=*/true)) {
+            return Status::Invalid(fmt::format(
+                "real-time store schema or mode mismatch for partition {}, bucket {}; recreate "
+                "the RealtimeContext",
+                PartitionToString(partition_bucket.partition), partition_bucket.bucket));
         }
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeReadView> read_view,
-                               iter->second->AcquireReadView());
+                               iter->second.store->AcquireReadView());
         if (!read_view) {
             return Status::Invalid("real-time store returned a null read view");
         }
@@ -117,27 +141,54 @@ Result<RealtimeStoreState> RealtimeContextImpl::GetOrCreateRealtimeStore(
                 initial_offset = memory_range->end;
             }
         }
-        return RealtimeStoreState{iter->second, initial_offset};
+        return RealtimeStoreState{iter->second.store, initial_offset};
+    }
+    if (!request.memory_pool) {
+        return Status::Invalid("real-time store memory pool is null");
+    }
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(
+        arrow::ExportSchema(*requested_schema, request.write_schema.get()));
+    RealtimeStoreMode mode = request.mode;
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeStore> store,
+                           factory_->Create(std::move(request)));
+    if (!store) {
+        return Status::Invalid("real-time store factory returned a null store");
     }
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<RealtimeStore> store,
-        factory_->Create(std::move(write_schema), statistics_mode, options, memory_pool));
-    stores_.emplace(key, store);
+    stores_.emplace(partition_bucket, StoreEntry{store, requested_schema, mode});
     if (offset_iter != committed_offsets_.end()) {
-        reclaimed_offsets_.emplace(key, offset_iter->second);
+        reclaimed_offsets_.emplace(partition_bucket, offset_iter->second);
     }
     return RealtimeStoreState{std::move(store), initial_offset};
 }
 
+Result<int64_t> RealtimeContextImpl::AdvanceMaterializedMaxSequenceNumber(
+    const RealtimePartitionBucket& partition_bucket, int64_t max_sequence_number) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    auto iter = stores_.find(partition_bucket);
+    if (iter == stores_.end()) {
+        return Status::KeyError(fmt::format("real-time store not found for partition {}, bucket {}",
+                                            PartitionToString(partition_bucket.partition),
+                                            partition_bucket.bucket));
+    }
+    StoreEntry& entry = iter->second;
+    if (max_sequence_number > entry.materialized_max_sequence_number) {
+        entry.materialized_max_sequence_number = max_sequence_number;
+    }
+    return entry.materialized_max_sequence_number;
+}
+
 Result<std::vector<RealtimePartitionBucketView>> RealtimeContextImpl::AcquireReadViews() {
     std::lock_guard<std::mutex> lock(mutex_);
     std::vector<RealtimePartitionBucketView> result;
     result.reserve(stores_.size());
     for (const auto& [partition_bucket, store] : stores_) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeReadView> read_view,
-                               store->AcquireReadView());
+                               store.store->AcquireReadView());
+        if (!read_view) {
+            return Status::Invalid("real-time store returned a null read view");
+        }
         result.push_back(
-            RealtimePartitionBucketView{partition_bucket, store, std::move(read_view)});
+            RealtimePartitionBucketView{partition_bucket, store.store, std::move(read_view)});
     }
     return result;
 }
@@ -270,7 +321,7 @@ Status RealtimeContextImpl::AdvanceCommittedProgress(int64_t snapshot_id,
             }
             auto store_iter = stores_.find(partition_bucket);
             if (store_iter != stores_.end()) {
-                notifications.emplace_back(partition_bucket, store_iter->second,
+                notifications.emplace_back(partition_bucket, store_iter->second.store,
                                            committed_end_offset);
             }
         }
diff --git a/src/paimon/core/realtime/realtime_context_impl.h b/src/paimon/core/realtime/realtime_context_impl.h
index 66c324cab..ea069a5cd 100644
--- a/src/paimon/core/realtime/realtime_context_impl.h
+++ b/src/paimon/core/realtime/realtime_context_impl.h
@@ -32,12 +32,16 @@
 #include <vector>
 
 #include "paimon/realtime/realtime_context.h"
+#include "paimon/realtime/realtime_store.h"
 #include "paimon/result.h"
-#include "paimon/statistics_mode.h"
 #include "paimon/visibility.h"
 
 struct ArrowSchema;
 
+namespace arrow {
+class Schema;
+}  // namespace arrow
+
 namespace paimon {
 
 class RealtimeStore;
@@ -66,10 +70,10 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
         const std::shared_ptr<RealtimeContext>& context);
 
     Result<RealtimeStoreState> GetOrCreateRealtimeStore(
-        const std::map<std::string, std::string>& partition, int32_t bucket,
-        std::unique_ptr<::ArrowSchema> write_schema, StatisticsMode statistics_mode,
-        const std::map<std::string, std::string>& options,
-        const std::shared_ptr<MemoryPool>& memory_pool);
+        RealtimeStoreCreateRequest&& request, const RealtimePartitionBucket& partition_bucket);
+
+    Result<int64_t> AdvanceMaterializedMaxSequenceNumber(
+        const RealtimePartitionBucket& partition_bucket, int64_t max_sequence_number);
 
     Result<std::vector<RealtimePartitionBucketView>> AcquireReadViews();
 
@@ -93,6 +97,13 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
         std::chrono::steady_clock::time_point expire_at;
     };
 
+    struct StoreEntry {
+        std::shared_ptr<RealtimeStore> store;
+        std::shared_ptr<arrow::Schema> write_schema;
+        RealtimeStoreMode mode;
+        int64_t materialized_max_sequence_number = -1;
+    };
+
     explicit RealtimeContextImpl(const std::shared_ptr<RealtimeStoreFactory>& factory);
 
     Status Start();
@@ -102,7 +113,7 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
     std::shared_ptr<RealtimeStoreFactory> factory_;
     std::mutex mutex_;
     std::mutex progress_mutex_;
-    std::map<RealtimePartitionBucket, std::shared_ptr<RealtimeStore>> stores_;
+    std::map<RealtimePartitionBucket, StoreEntry> stores_;
     // Full-table progress used as the initial offset when a store is created lazily.
     RealtimeOffsetMap committed_offsets_;
     // Progress already reflected in stores owned by this context.
diff --git a/src/paimon/core/realtime/realtime_context_test.cpp b/src/paimon/core/realtime/realtime_context_test.cpp
index 017820fd4..c9531eda5 100644
--- a/src/paimon/core/realtime/realtime_context_test.cpp
+++ b/src/paimon/core/realtime/realtime_context_test.cpp
@@ -9,12 +9,11 @@
  *
  *   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.
+ * 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.
  */
 
 #include <chrono>
@@ -31,7 +30,6 @@
 #include "arrow/c/helpers.h"
 #include "paimon/core/realtime/realtime_context_impl.h"
 #include "paimon/memory/memory_pool.h"
-#include "paimon/realtime/realtime_store.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -49,26 +47,24 @@ class TestingRealtimeStore : public RealtimeStore {
     Status Write(RealtimeWriteBatch&&) override {
         return Status::OK();
     }
-
     Result<std::optional<std::shared_ptr<RealtimeSegmentHandle>>> SealForCommit() override {
         return std::optional<std::shared_ptr<RealtimeSegmentHandle>>();
     }
-
     Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
         const std::shared_ptr<RealtimeSegmentHandle>&) override {
         return std::vector<std::unique_ptr<BatchReader>>();
     }
-
     Result<std::shared_ptr<RealtimeReadView>> AcquireReadView() override {
         ++acquire_count;
+        if (return_null_read_view) {
+            return std::shared_ptr<RealtimeReadView>();
+        }
         return std::make_shared<TestingReadView>();
     }
-
     Result<std::vector<std::unique_ptr<BatchReader>>> CreateQueryReaders(
         const std::shared_ptr<RealtimeReadView>&, int64_t, const RealtimeQueryContext&) override {
         return std::vector<std::unique_ptr<BatchReader>>();
     }
-
     Status AdvanceCommittedOffset(int64_t committed_offset) override {
         ++advance_count;
         if (fail_next_advance) {
@@ -78,7 +74,6 @@ class TestingRealtimeStore : public RealtimeStore {
         committed_offsets.push_back(committed_offset);
         return Status::OK();
     }
-
     uint64_t GetMemoryUsage() const override {
         return 0;
     }
@@ -86,33 +81,37 @@ class TestingRealtimeStore : public RealtimeStore {
     int32_t acquire_count = 0;
     int32_t advance_count = 0;
     bool fail_next_advance = false;
+    bool return_null_read_view = false;
     std::vector<int64_t> committed_offsets;
 };
 
 class TestingRealtimeStoreFactory : public RealtimeStoreFactory {
  public:
-    Result<std::shared_ptr<RealtimeStore>> Create(std::unique_ptr<ArrowSchema> write_schema,
-                                                  StatisticsMode,
-                                                  const std::map<std::string, std::string>&,
-                                                  const std::shared_ptr<MemoryPool>&) override {
-        if (!write_schema || !write_schema->release) {
+    Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) override {
+        if (!request.write_schema || !request.write_schema->release) {
             return Status::Invalid("testing write schema is null");
         }
-        ArrowSchemaRelease(write_schema.get());
+        ArrowSchemaRelease(request.write_schema.get());
+        if (return_null_store) {
+            return std::shared_ptr<RealtimeStore>();
+        }
         auto store = std::make_shared<TestingRealtimeStore>();
         stores.push_back(store);
         return store;
     }
 
+    bool return_null_store = false;
     std::vector<std::shared_ptr<TestingRealtimeStore>> stores;
 };
 
-std::unique_ptr<ArrowSchema> MakeWriteSchema() {
-    auto c_schema = std::make_unique<ArrowSchema>();
+std::unique_ptr<ArrowSchema> MakeWriteSchema(
+    const std::shared_ptr<arrow::DataType>& id_type = arrow::int64(),
+    const std::shared_ptr<arrow::KeyValueMetadata>& metadata = nullptr) {
+    auto schema = std::make_unique<ArrowSchema>();
     EXPECT_TRUE(
-        arrow::ExportSchema(*arrow::schema({arrow::field("id", arrow::int64())}), c_schema.get())
+        arrow::ExportSchema(*arrow::schema({arrow::field("id", id_type)}, metadata), schema.get())
             .ok());
-    return c_schema;
+    return schema;
 }
 
 Result<std::shared_ptr<RealtimeContextImpl>> CreateContext(
@@ -122,35 +121,42 @@ Result<std::shared_ptr<RealtimeContextImpl>> CreateContext(
     return RealtimeContextImpl::Cast(context);
 }
 
+Result<RealtimeStoreState> GetOrCreateAppendStore(
+    const std::shared_ptr<RealtimeContextImpl>& context,
+    const std::map<std::string, 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,
+    StatisticsMode statistics_mode = StatisticsMode::NONE) {
+    return context->GetOrCreateRealtimeStore(
+        RealtimeStoreCreateRequest{std::move(write_schema), options, memory_pool,
+                                   RealtimeStoreMode::APPEND_ONLY, statistics_mode},
+        RealtimePartitionBucket(partition, bucket));
+}
+
 TEST(RealtimeContextTest, TestReusesStoreAndCapturesRegisteredViews) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    std::shared_ptr<MemoryPool> pool = GetDefaultPool();
-
+    ASSERT_OK_AND_ASSIGN(RealtimeStoreState first,
+                         GetOrCreateAppendStore(context, {{"dt", "2026-08-02"}}, 0,
+                                                MakeWriteSchema(), {{"k", "v"}}, GetDefaultPool()));
+    ASSERT_EQ(0, first.initial_offset);
     ASSERT_OK_AND_ASSIGN(
-        RealtimeStoreState first_state,
-        context->GetOrCreateRealtimeStore({{"dt", "2026-08-02"}}, 0, MakeWriteSchema(),
-                                          StatisticsMode::NONE, {{"k", "v"}}, pool));
-    ASSERT_EQ(0, first_state.initial_offset);
-    ASSERT_OK_AND_ASSIGN(
-        RealtimeStoreState first_again_state,
-        context->GetOrCreateRealtimeStore({{"dt", "2026-08-02"}}, 0, MakeWriteSchema(),
-                                          StatisticsMode::NONE, {}, pool));
-    ASSERT_EQ(first_state.store, first_again_state.store);
-    ASSERT_EQ(0, first_again_state.initial_offset);
+        RealtimeStoreState second,
+        GetOrCreateAppendStore(context, {{"dt", "2026-08-02"}}, 0, MakeWriteSchema(), {},
+                               GetDefaultPool(), StatisticsMode::FULL));
+    ASSERT_EQ(first.store, second.store);
+    ASSERT_EQ(0, second.initial_offset);
     ASSERT_EQ(1, factory->stores.size());
     ASSERT_EQ(1, factory->stores[0]->acquire_count);
 
-    ASSERT_OK_AND_ASSIGN(
-        RealtimeStoreState second_state,
-        context->GetOrCreateRealtimeStore({{"dt", "2026-08-02"}}, 1, MakeWriteSchema(),
-                                          StatisticsMode::NONE, {}, pool));
-    ASSERT_OK_AND_ASSIGN(
-        RealtimeStoreState third_state,
-        context->GetOrCreateRealtimeStore({{"dt", "2026-08-03"}}, 0, MakeWriteSchema(),
-                                          StatisticsMode::NONE, {}, pool));
-    ASSERT_NE(first_state.store, second_state.store);
-    ASSERT_NE(first_state.store, third_state.store);
+    ASSERT_OK_AND_ASSIGN(RealtimeStoreState third,
+                         GetOrCreateAppendStore(context, {{"dt", "2026-08-02"}}, 1,
+                                                MakeWriteSchema(), {}, GetDefaultPool()));
+    ASSERT_OK_AND_ASSIGN(RealtimeStoreState fourth,
+                         GetOrCreateAppendStore(context, {{"dt", "2026-08-03"}}, 0,
+                                                MakeWriteSchema(), {}, GetDefaultPool()));
+    ASSERT_NE(first.store, third.store);
+    ASSERT_NE(first.store, fourth.store);
     ASSERT_EQ(3, factory->stores.size());
 
     ASSERT_OK_AND_ASSIGN(std::vector<RealtimePartitionBucketView> views,
@@ -158,23 +164,87 @@ TEST(RealtimeContextTest, TestReusesStoreAndCapturesRegisteredViews) {
     ASSERT_EQ(3, views.size());
     const RealtimePartitionBucket expected_partition_bucket({{"dt", "2026-08-02"}}, 0);
     ASSERT_EQ(expected_partition_bucket, views[0].partition_bucket);
-    ASSERT_EQ(first_state.store, views[0].store);
+    ASSERT_EQ(first.store, views[0].store);
     ASSERT_TRUE(views[0].read_view);
     ASSERT_EQ(2, factory->stores[0]->acquire_count);
     ASSERT_EQ(1, factory->stores[1]->acquire_count);
     ASSERT_EQ(1, factory->stores[2]->acquire_count);
 }
 
+TEST(RealtimeContextTest, TestRejectsMismatchedModeOnStoreReuse) {
+    auto factory = std::make_shared<TestingRealtimeStoreFactory>();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
+    const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(), {}, GetDefaultPool()));
+
+    ASSERT_NOK_WITH_MSG(
+        context->GetOrCreateRealtimeStore(
+            RealtimeStoreCreateRequest{
+                MakeWriteSchema(), {}, GetDefaultPool(), RealtimeStoreMode::PRIMARY_KEY},
+            RealtimePartitionBucket(partition, 0)),
+        "schema or mode mismatch for partition {dt=2026-08-02}, bucket 0; recreate the "
+        "RealtimeContext");
+    ASSERT_EQ(1, factory->stores.size());
+}
+
+TEST(RealtimeContextTest, TestRejectsMismatchedSchemaOnStoreReuse) {
+    auto factory = std::make_shared<TestingRealtimeStoreFactory>();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
+    const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
+    std::shared_ptr<arrow::KeyValueMetadata> metadata =
+        arrow::key_value_metadata({"identity"}, {"v1"});
+    ASSERT_OK(GetOrCreateAppendStore(
+        context, partition, 0, MakeWriteSchema(arrow::int64(), metadata), {}, GetDefaultPool()));
+    ASSERT_NOK_WITH_MSG(
+        GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(arrow::int32(), metadata), {},
+                               GetDefaultPool()),
+        "schema or mode mismatch for partition {dt=2026-08-02}, bucket 0; recreate the "
+        "RealtimeContext");
+    ASSERT_NOK_WITH_MSG(
+        GetOrCreateAppendStore(
+            context, partition, 0,
+            MakeWriteSchema(arrow::int64(), arrow::key_value_metadata({"identity"}, {"v2"})), {},
+            GetDefaultPool()),
+        "schema or mode mismatch for partition {dt=2026-08-02}, bucket 0; recreate the "
+        "RealtimeContext");
+    ASSERT_EQ(1, factory->stores.size());
+}
+
+TEST(RealtimeContextTest, TestReconcilesPrimaryKeyInitialSequence) {
+    auto factory = std::make_shared<TestingRealtimeStoreFactory>();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
+    const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
+    const RealtimePartitionBucket partition_bucket(partition, /*bucket=*/0);
+
+    ASSERT_OK(context->GetOrCreateRealtimeStore(
+        RealtimeStoreCreateRequest{MakeWriteSchema(), /*options=*/{}, GetDefaultPool(),
+                                   RealtimeStoreMode::PRIMARY_KEY},
+        partition_bucket));
+
+    ASSERT_OK_AND_ASSIGN(int64_t first, context->AdvanceMaterializedMaxSequenceNumber(
+                                            partition_bucket, /*max_sequence_number=*/4));
+    ASSERT_EQ(4, first);
+    ASSERT_OK_AND_ASSIGN(int64_t second, context->AdvanceMaterializedMaxSequenceNumber(
+                                             partition_bucket, /*max_sequence_number=*/8));
+    ASSERT_EQ(8, second);
+    ASSERT_OK_AND_ASSIGN(int64_t third, context->AdvanceMaterializedMaxSequenceNumber(
+                                            partition_bucket, /*max_sequence_number=*/6));
+    ASSERT_EQ(8, third);
+    ASSERT_OK_AND_ASSIGN(int64_t fourth, context->AdvanceMaterializedMaxSequenceNumber(
+                                             partition_bucket, /*max_sequence_number=*/10));
+    ASSERT_EQ(10, fourth);
+}
+
 TEST(RealtimeContextTest, TestCommittedProgressIsMonotonicAndSelective) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
 
-    ASSERT_OK(context->GetOrCreateRealtimeStore(partition, 0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(partition, 1, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(), {}, GetDefaultPool()));
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 1, MakeWriteSchema(), {}, GetDefaultPool()));
     ASSERT_EQ(2, factory->stores.size());
 
     ASSERT_NOK_WITH_MSG(context->AdvanceCommittedProgress(-1, {}),
@@ -191,10 +261,9 @@ TEST(RealtimeContextTest, TestCommittedProgressIsMonotonicAndSelective) {
     ASSERT_EQ(std::vector<int64_t>({7}), factory->stores[0]->committed_offsets);
     ASSERT_TRUE(factory->stores[1]->committed_offsets.empty());
 
-    ASSERT_OK_AND_ASSIGN(
-        RealtimeStoreState restored_state,
-        context->GetOrCreateRealtimeStore({{"dt", "unknown"}}, 0, MakeWriteSchema(),
-                                          StatisticsMode::NONE, {}, pool));
+    ASSERT_OK_AND_ASSIGN(RealtimeStoreState restored_state,
+                         GetOrCreateAppendStore(context, {{"dt", "unknown"}}, 0, MakeWriteSchema(),
+                                                {}, GetDefaultPool()));
     ASSERT_EQ(9, restored_state.initial_offset);
 
     ASSERT_OK(context->AdvanceCommittedProgress(
@@ -214,7 +283,6 @@ TEST(RealtimeContextTest, TestCommittedProgressIsMonotonicAndSelective) {
 TEST(RealtimeContextTest, TestRemovedInactivePartitionDoesNotRequireReopen) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     const std::map<std::string, std::string> active_partition = {{"dt", "2026-08-02"}};
     const std::map<std::string, std::string> inactive_partition = {{"dt", "2026-08-03"}};
     const RealtimePartitionBucket active_partition_bucket(active_partition, /*bucket=*/0);
@@ -223,29 +291,28 @@ TEST(RealtimeContextTest, TestRemovedInactivePartitionDoesNotRequireReopen) {
     ASSERT_OK(context->AdvanceCommittedProgress(
         5, {{active_partition_bucket, /*offset=*/7}, {inactive_partition_bucket, /*offset=*/9}}));
     ASSERT_OK_AND_ASSIGN(RealtimeStoreState active_state,
-                         context->GetOrCreateRealtimeStore(active_partition, 0, MakeWriteSchema(),
-                                                           StatisticsMode::NONE, {}, pool));
+                         GetOrCreateAppendStore(context, active_partition, 0, MakeWriteSchema(), {},
+                                                GetDefaultPool()));
     ASSERT_EQ(7, active_state.initial_offset);
 
     ASSERT_OK(context->AdvanceCommittedProgress(6, {{active_partition_bucket, /*offset=*/7}}));
     ASSERT_OK_AND_ASSIGN(RealtimeStoreState inactive_state,
-                         context->GetOrCreateRealtimeStore(inactive_partition, 0, MakeWriteSchema(),
-                                                           StatisticsMode::NONE, {}, pool));
+                         GetOrCreateAppendStore(context, inactive_partition, 0, MakeWriteSchema(),
+                                                {}, GetDefaultPool()));
     ASSERT_EQ(0, inactive_state.initial_offset);
 }
 
 TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
 
-    ASSERT_OK(context->GetOrCreateRealtimeStore(partition, 0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(partition, 1, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(partition, 2, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(), {}, GetDefaultPool()));
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 1, MakeWriteSchema(), {}, GetDefaultPool()));
+    ASSERT_OK(
+        GetOrCreateAppendStore(context, partition, 2, MakeWriteSchema(), {}, GetDefaultPool()));
     ASSERT_EQ(3, factory->stores.size());
     factory->stores[1]->fail_next_advance = true;
 
@@ -259,9 +326,9 @@ TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
     ASSERT_TRUE(factory->stores[1]->committed_offsets.empty());
     ASSERT_EQ(std::vector<int64_t>({9}), factory->stores[2]->committed_offsets);
 
-    ASSERT_OK_AND_ASSIGN(RealtimeStoreState failed_store_state,
-                         context->GetOrCreateRealtimeStore(partition, 1, MakeWriteSchema(),
-                                                           StatisticsMode::NONE, {}, pool));
+    ASSERT_OK_AND_ASSIGN(
+        RealtimeStoreState failed_store_state,
+        GetOrCreateAppendStore(context, partition, 1, MakeWriteSchema(), {}, GetDefaultPool()));
     ASSERT_EQ(8, failed_store_state.initial_offset);
 
     ASSERT_OK(context->AdvanceCommittedProgress(5, committed_offsets));
@@ -274,16 +341,15 @@ TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
 TEST(RealtimeContextTest, TestRequiresReopenWhenCommittedProgressMovesBackwards) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     const std::map<std::string, std::string> first_partition = {{"dt", "2026-08-02"}};
     const std::map<std::string, std::string> second_partition = {{"dt", "2026-08-03"}};
     const RealtimePartitionBucket first_partition_bucket(first_partition, /*bucket=*/0);
     const RealtimePartitionBucket second_partition_bucket(second_partition, /*bucket=*/0);
 
-    ASSERT_OK(context->GetOrCreateRealtimeStore(first_partition, 0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(second_partition, 0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, pool));
+    ASSERT_OK(GetOrCreateAppendStore(context, first_partition, 0, MakeWriteSchema(), {},
+                                     GetDefaultPool()));
+    ASSERT_OK(GetOrCreateAppendStore(context, second_partition, 0, MakeWriteSchema(), {},
+                                     GetDefaultPool()));
     ASSERT_OK(context->AdvanceCommittedProgress(
         5, {{first_partition_bucket, /*offset=*/7}, {second_partition_bucket, /*offset=*/9}}));
     ASSERT_EQ(std::vector<int64_t>({7}), factory->stores[0]->committed_offsets);
@@ -308,8 +374,8 @@ TEST(RealtimeContextTest, TestRequiresReopenWhenCommittedProgressMovesBackwards)
 TEST(RealtimeContextTest, TestPinsResolvesAndReleasesReadViewTicket) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(/*partition=*/{}, /*bucket=*/0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, GetDefaultPool()));
+    ASSERT_OK(GetOrCreateAppendStore(context, /*partition=*/{}, /*bucket=*/0, MakeWriteSchema(), {},
+                                     GetDefaultPool()));
     ASSERT_OK_AND_ASSIGN(std::vector<RealtimePartitionBucketView> views,
                          context->AcquireReadViews());
     ASSERT_EQ(1, views.size());
@@ -332,8 +398,8 @@ TEST(RealtimeContextTest, TestPinsResolvesAndReleasesReadViewTicket) {
 TEST(RealtimeContextTest, TestExpiresAbandonedReadViewTicket) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
-    ASSERT_OK(context->GetOrCreateRealtimeStore(/*partition=*/{}, /*bucket=*/0, MakeWriteSchema(),
-                                                StatisticsMode::NONE, {}, GetDefaultPool()));
+    ASSERT_OK(GetOrCreateAppendStore(context, /*partition=*/{}, /*bucket=*/0, MakeWriteSchema(), {},
+                                     GetDefaultPool()));
     ASSERT_OK_AND_ASSIGN(std::vector<RealtimePartitionBucketView> views,
                          context->AcquireReadViews());
     ASSERT_EQ(1, views.size());
@@ -354,5 +420,20 @@ TEST(RealtimeContextTest, TestRejectsNullFactory) {
                         "real-time store factory is null");
 }
 
+TEST(RealtimeContextTest, TestRejectsNullPluginResults) {
+    auto factory = std::make_shared<TestingRealtimeStoreFactory>();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
+    factory->return_null_store = true;
+    ASSERT_NOK_WITH_MSG(GetOrCreateAppendStore(context, /*partition=*/{}, /*bucket=*/0,
+                                               MakeWriteSchema(), {}, GetDefaultPool()),
+                        "real-time store factory returned a null store");
+
+    factory->return_null_store = false;
+    ASSERT_OK(GetOrCreateAppendStore(context, /*partition=*/{}, /*bucket=*/0, MakeWriteSchema(), {},
+                                     GetDefaultPool()));
+    factory->stores[0]->return_null_read_view = true;
+    ASSERT_NOK_WITH_MSG(context->AcquireReadViews(), "real-time store returned a null read view");
+}
+
 }  // namespace
 }  // namespace paimon::test