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

[HaHaJeff]

Purpose

Linked issue: #158

This PR extends the pluggable realtime read and write support introduced by #163 and the RealtimeStore API from #199 to fixed-bucket primary-key tables.

Applications attach a RealtimeContext to the existing file-store paths. Its RealtimeStoreFactory receives a RealtimeStoreCreateRequest with a RealtimeStoreMode and creates an append-only or primary-key store. The built-in primary-key store is storage-oriented: it retains prepared mutations and returns one reader per prepared batch. The framework owns offset and sequence preparation, sorting, visibility filtering, merge-on-read, and normal data-file writing.

For primary-key reads, immutable memory read views are combined with the selected disk snapshot through RealtimeSplit. The store applies the RealtimeQueryContext::read_schema projection to memory batches, including nested field-ID alignment within the current table schema. Schema evolution requires recreating the realtime context and store.

All disk splits for a partition-bucket are folded into one RealtimeSplit. Overlapping runs are merged within each disk section, and the non-overlapping sections are concatenated into one sorted disk reader before the final disk-plus-memory merge-on-read. This keeps disk-side final merge fan-in bounded while ensuring that every disk run and memory mutation participates in one primary-key merge. Predicates that are unsafe before merge are evaluated after merge.

Memory-side reader fan-in is not hard-bounded: every prepared memory batch still contributes one reader to the final merge. Many small writes therefore increase final merge fan-in and the number of retained first batches. This is a current performance and resource limitation, not a correctness problem.

The main changes are:

• add an in-memory primary-key RealtimeStore implementation;
• let factories dispatch on RealtimeStoreCreateRequest::mode;
• let MergeTreeWriter consume prepared readers returned by the primary-key store;
• reuse the existing key-value reader and merge path for disk and memory data;
• restore offsets from snapshot metadata and sequence numbers from committed bucket files;
• reuse the realtime file-store and two-stage scan/read APIs introduced by feat: support real-time append writes and reads with pluggable memory indexers #163 and feat(realtime): support two-stage scan and read #199; and
• preserve non-realtime and append-realtime behavior.

The current built-in primary-key implementation supports fixed-bucket tables with the deduplicate merge engine, full-row mutations, latest-snapshot recovery, concurrent readers, and synchronized writer operations. The supported lifecycle uses one active writer with its realtime context. Calls coordinating write, prepare-commit, commit, refresh, and reads may run concurrently as covered by the integration tests; writer handoff is sequential, after the prior writer is closed, with progress restored from the committed snapshot.

Dynamic buckets, lookup or early merge-on-read, aggregation and partial-update merge engines, data evolution within an existing context/store, user sequence fields, read-optimized scans, ignore_previous_files, custom write schemas, and recovery from a non-latest snapshot are not included.

Writer-local compaction is force-disabled for realtime primary-key writers. User-provided compaction options are ignored, num-sorted-run.stop-trigger backpressure does not apply, and level-0 runs accumulate with each commit, so external compaction is required.

The built-in primary-key store keeps realtime mutations entirely in memory and does not implement spill. write-buffer-size does not apply to this path, so memory usage is bounded only by how much the caller writes before prepare commit. The public RealtimeStore contract still permits custom implementations to use their own spill strategy.

Tests

Added unit coverage for:

• factory creation and mode dispatch;
• primary-key write validation, sealing, and prepared transport-schema checks;
• sequence ordering, row kinds, deduplication, and commit readers;
• one reader per prepared batch and immutable read-view offset ranges;
• full trimmed primary-key projection when key columns are omitted, including composite keys;
• nested field-ID projection alignment within the current schema;
• disk-section composition before final disk-plus-memory merge;
• realtime offset progress and snapshot refresh;
• sorted-reader writes in MergeTreeWriter with multiple readers, overlapping keys, and duplicate-key deduplication; and
• supported option validation, including rejection of floating-point primary keys and enabled global index.

Added integration coverage for:

• primary-key realtime write, prepare, commit, refresh, and reopen;
• latest-snapshot recovery of offsets and sequence numbers;
• merge-on-read across committed files and memory segments, including all disk splits of a bucket merged with memory rows;
• deletes, repeated keys, keyless and nested projection, predicates, external compaction, and sequential writer handoff;
• concurrent write, prepare-commit, commit, refresh, and read operations on one active writer/context lifecycle; and
• non-realtime and append-realtime regression paths.

Integration tests write real ORC data files through the normal FileStoreWrite, PrepareCommitWithProgress, and CommitWithProgress paths, then read the data back from the committed snapshot without a RealtimeContext.

The focused primary-key realtime tests pass under ASAN, UBSAN, and LeakSanitizer. The final cleanup was unchanged from the already validated code behavior: both relevant targets built, 39 focused core tests passed, and all 57 realtime integration tests passed.

API and Format

This PR reuses the public realtime file-store APIs introduced by #163 and #199, including RealtimeContext, RealtimeWriteBatch, PrepareCommitWithProgress, CommitWithProgress, realtime split planning, and snapshot refresh. It does not add a separate primary-key table API.

Factories implement RealtimeStoreFactory::Create(RealtimeStoreCreateRequest&&). The request carries the write or prepared transport schema, options, memory pool, statistics mode, and RealtimeStoreMode; factories dispatch on the mode. For primary-key mode, the framework supplies batches with its prepared transport schema and assigns offsets and sequence numbers before calling the store.

The store returns raw prepared readers. For primary-key queries, offset_begin is ignored by the store; the framework applies visibility filtering, final projection and predicates, and merge-on-read after reader creation. RealtimePrimaryKeyLayout::CreateSchema and RealtimePrimaryKeyLayout::ValidateSchema, together with the layout indexes, define the transport schema used by write, commit, and query paths.

No new data-file or commit-message format is introduced. Primary-key realtime writes produce normal merge-tree data files and commit messages. Realtime offsets continue to use the versioned snapshot metadata introduced by #163; they are framework-assigned progress identifiers, not primary-key sequence numbers.

Paimon serializes Write and SealForCommit for each store. Existing immutable read views remain valid across later writes, seals, refresh, and committed-offset reclamation. Realtime split tickets remain process-local and single-success-use as defined by #199.

After a write, prepare-commit, commit, or refresh failure, the caller discards the writer and realtime context, recreates both from the latest committed snapshot, and replays its external WAL. The failed writer, store, and context must not be reused because retained or sealed memory may have unknown durability. Overwrite, truncate, partition drop, rollback, and other progress-resetting operations likewise require coordinated recreation before writes continue.

Existing non-realtime tables and append-realtime tables retain their previous execution paths.

Documentation

The public headers document store creation and mode dispatch, prepared transport schemas, offset and sequence separation, read-schema projection, ownership, concurrency, and immutable-view behavior. The limitations and failure-recovery contract above describe the supported current implementation.

Generative AI tooling

Codex (GPT-5) was used for implementation, refactoring, tests, and PR text. Claude Code (Claude Opus 4.8) was used for review.

[wangyong9999]

Two correctness issues found in the primary-key realtime path.

[wangyong9999]

Only the watermark returned to the new MergeTreeWriter is advanced here. When stores_ already contains this partition-bucket, PrimaryKeyRealtimeStore::next_sequence_number_ remains on the old watermark. If the latest snapshot has advanced from sequence 4 to 10, the next in-memory mutation can get 5 while the same row gets 11 when flushed; merge-on-read then lets the disk row at 10 hide the newer memory row. Advance or reject the reused store atomically, or use one shared sequence allocator for both paths.

[HaHaJeff]

Fixed in 5f74e46. Reusing an existing PK realtime store now rejects a restore watermark above its materialized watermark, preventing the store and writer sequence allocators from diverging; the context test covers this case.

[HaHaJeff]

Thanks for catching this. The original implementation let the store assign sequence numbers because it also performed the PK merge. After moving preparation and MOR into the framework, keeping another allocator in the store would create two sequence authorities and make writer handoff unsafe. Commit 141099c moved the partition-bucket materialized sequence watermark into RealtimeContext; replacement writers initialize from it and advance it only after a successful store write. V1 follows the same lifecycle contract as append realtime: one RealtimeContext is owned by one active FileStoreWrite, so simultaneous writers sharing a context are unsupported.

[wangyong9999]

This maps nested projections only at the top level. With stored payload<a,b> and requested payload<b>, KeyValueProjectionReader builds the pruned struct but reads child 0 from the full stored row, returning a as b when their types match. Align each stored batch to the requested nested type—as ArrowRealtimeStore does with AlignArrayToReadType—before building the KeyValue reader, and add a memory-plus-disk nested-projection test.

[HaHaJeff]

Fixed in 5f74e46, with schema-evolution and field-ID follow-ups in 43afe02 and 9d2d19c. Stored PK batches are aligned to the requested nested type before constructing the KeyValue reader, and the added unit/integration coverage verifies payload across memory and disk.

[HaHaJeff]

Addressed at current PR head a7bc03b7b23c0302ad77d0e7312c0fb4da8a334c. PrimaryKeyRealtimeStore::CreateQueryReaders now aligns each stored batch to RealtimeQueryContext::read_schema through NestedProjectionUtils::AlignArrayToReadType, matching the append-store query path. PreparedKeyValueReader only converts the already projected transport schema; the earlier framework-side recursive schema-reconciliation path has been removed. Schema evolution remains unsupported in V1 and requires recreating the RealtimeContext/store. TestPkNestedProjectionAcrossDiskAndMemory covers disk, sealed memory, and active memory with nested projection.

[lxy-9602]

Thank you for the contribution. The code looks clear and well organized. Before diving into the detailed review, I would like to discuss two design points.

First, it seems that internal sequence-number assignment and per-batch primary-key sorting are currently handled inside the realtime store implementation. I suggest moving these responsibilities into the Paimon framework instead.

The framework could assign offsets and sequence numbers, append internal fields such as _VALUE_KIND, _SEQUENCE_NUMBER, and _REALTIME_OFFSET, and physically sort each input Arrow batch before passing it to the store plugin. The plugin would then only manage storage concerns, without needing to understand PK sorting rules, sequence fields, or merge-engine semantics.

Query and prepare-commit could convert these already sorted batches into KeyValueRecordReaders and reuse the existing SortMergeReader and merge functions. The flush path could also accept sorted readers directly, avoiding sequence reassignment and repeated per-batch sorting. This would make custom plugins easier to implement and allow realtime reads and writes to reuse the framework’s existing merge-engine and sequence-field behavior.

I think this can be the first-stage solution. If profiling later shows that copying data to produce physically sorted Arrow batches is a real write-path bottleneck, we could introduce a shallow-copy mode based on sorted indices. That would require significantly more interface changes, so I suggest optimizing it only after it becomes an observed hotspot.

Second, the in-memory store could keep PK statistics for each batch, such as min/max values. Predicates on value fields may not be pushable, but predicate_for_keys should be applicable to these statistics so irrelevant in-memory batches can be pruned during reads. This optimization could also be implemented in a follow-up PR.

[HaHaJeff]

Thank you for the contribution. The code looks clear and well organized. Before diving into the detailed review, I would like to discuss two design points.

First, it seems that internal sequence-number assignment and per-batch primary-key sorting are currently handled inside the realtime store implementation. I suggest moving these responsibilities into the Paimon framework instead.

The framework could assign offsets and sequence numbers, append internal fields such as _VALUE_KIND, _SEQUENCE_NUMBER, and _REALTIME_OFFSET, and physically sort each input Arrow batch before passing it to the store plugin. The plugin would then only manage storage concerns, without needing to understand PK sorting rules, sequence fields, or merge-engine semantics.

Query and prepare-commit could convert these already sorted batches into KeyValueRecordReaders and reuse the existing SortMergeReader and merge functions. The flush path could also accept sorted readers directly, avoiding sequence reassignment and repeated per-batch sorting. This would make custom plugins easier to implement and allow realtime reads and writes to reuse the framework’s existing merge-engine and sequence-field behavior.

I think this can be the first-stage solution. If profiling later shows that copying data to produce physically sorted Arrow batches is a real write-path bottleneck, we could introduce a shallow-copy mode based on sorted indices. That would require significantly more interface changes, so I suggest optimizing it only after it becomes an observed hotspot.

Second, the in-memory store could keep PK statistics for each batch, such as min/max values. Predicates on value fields may not be pushable, but predicate_for_keys should be applicable to these statistics so irrelevant in-memory batches can be pruned during reads. This optimization could also be implemented in a follow-up PR.

Thanks for the detailed suggestion. I agree that sequence assignment, PK sorting, and merge semantics should belong to the Paimon framework rather than the real-time store plugin.

The current implementation assigns sequence numbers and performs PK sorting and in-memory merging inside the PK store. During prepare-commit, it converts the returned batches back into ordinary RecordBatches and passes them through WriteBuffer, which assigns sequence numbers and sorts the same data again. I plan to revise this design as follows.

Framework-side batch preparation

Before calling RealtimeStore::Write, the Paimon framework will:

1. assign _REALTIME_OFFSET and _SEQUENCE_NUMBER atomically according to the original per-row write order;
2. materialize _VALUE_KIND, _SEQUENCE_NUMBER, and _REALTIME_OFFSET;
3. physically and stably sort the complete Arrow batch by primary key.

All columns will be reordered with the same sort indices, so the value, row kind, sequence number, and real-time offset remain associated with the same mutation.

Sorting will not perform deduplication or early MOR. Every mutation will remain in the prepared batch. The progress counters will advance only after RealtimeStore::Write succeeds.

RealtimeStore responsibility

RealtimeStore will treat the internal fields as opaque Arrow columns and preserve the prepared batches through write, seal, read-view, query-reader, and commit-reader operations.

It will no longer:

• assign sequence numbers;
• understand PK sorting rules;
• depend on merge functions or merge-engine semantics;
• perform PK deduplication or MOR.

Each physically sorted input batch will represent one independent sorted run. A store may return multiple readers, and Paimon will merge those runs in the framework. The built-in and custom stores will therefore use the same path.

Query path

The framework will provide an adapter from the store's BatchReader to KeyValueRecordReader.

For PK queries:

1. the store returns the prepared sorted batches;
2. the adapter uses _REALTIME_OFFSET to remove memory rows already covered by the selected snapshot;
3. the adapter converts the remaining rows into sorted KeyValueRecordReaders;
4. the existing SortMergeReader merges the memory readers with disk readers;
5. the existing merge function performs MOR.

_SEQUENCE_NUMBER remains the row-version field used to resolve versions during disk-memory MOR.

Prepare-commit path

RealtimeStore and MergeTreeWriter will not depend on each other directly. The framework-owned RealtimePrimaryKeyWriter will coordinate them:

1. call RealtimeStore::SealForCommit to obtain an immutable segment;
2. call RealtimeStore::CreateCommitReaders for that segment;
3. adapt the returned BatchReaders into sorted KeyValueRecordReaders;
4. pass those readers to MergeTreeWriter::WriteSortedReaders;
5. call the existing MergeTreeWriter::PrepareCommit;
6. attach the sealed segment's real-time progress to the resulting commit progress.

The resulting path will be:

RealtimeStore
  -> BatchReader
  -> framework BatchReader-to-KeyValueRecordReader adapter
  -> MergeTreeWriter::WriteSortedReaders
  -> existing SortMergeReader and merge functions
  -> existing rolling data-file writer
  -> CommitIncrement

[lxy-9602]

Thank you for your response! The current direction looks good to me. @zjw1111 , could you also take a look?

Also, the offset filtering for PK tables has now been moved to the framework layer, while for append tables it is still handled inside the plugin through the offset_begin parameter in CreateQueryReaders. I plan to align the append-table path later as well, similar to PK tables, by moving the offset filtering into the framework layer. For this PR, I think it’s fine to keep the current interface for now and focus on implementing the PK-table part first.

[zjw1111]

Thank you for your response! The current direction looks good to me. @zjw1111 , could you also take a look?

LGTM

[HaHaJeff]

Thanks for the design guidance. The initial implementation placed sequence assignment, sorting, and in-memory MOR in the PK store, which made custom stores understand Paimon merge semantics and caused data to be sorted again during file writing. Commits 8d96153, df322c1, and 87e4548 moved sorted-reader writing, reader adaptation, transport-field materialization, and PK sorting into the framework. Commit ff44419 finalized the PK storage-only boundary by removing store-side PK merge dependencies. The framework allocates sequence and offsets under the V1 one-context/one-active-writer contract. PK statistics pruning remains follow-up work; predicate pruning before MOR stays disabled for correctness.

[wangyong9999]

This is the only ValidateFull() in the non-test tree, and at this point it cannot protect anything: PrepareBatch already ran SortIndices and Take over these same buffers before exporting the array, so a malformed input would have been dereferenced well before we get here. What is left is an O(rows) pass (offset walks, UTF-8 checks) on every write, over an array Paimon itself just built. ArrowRealtimeStore::Write does nothing comparable.

Suggest dropping it. If input validation is wanted, it belongs right after ImportArray in PrepareBatch, before any kernel touches the data.

[HaHaJeff]

Thanks, agreed. The redundant ValidateFull() was removed in ff44419. The built-in store now imports and retains the framework-prepared batch without another full-array validation pass after the framework kernels have consumed it.

[wangyong9999]

Two costs stack up here on every batch of both the commit and the query path:

• the key projection plus ColumnarBatchContext is built here, and then built again for key_ctx_ in NextBatchImpl; on the commit path visible_offsets_ is empty, so ApplyOffsetFilter is a no-op and the two are over the identical array.
• the per-row CompareTo repeats the comparison that the downstream MergedKeyValueRecordReader is about to do on the same adjacent pairs.

At minimum, build the key context once and hand it to key_ctx_. Beyond that, for the built-in store this verifies Paimon's own PrepareBatch output, so making it a debug-only check (or applying it only to third-party stores) would keep the plugin-contract value without paying for it on every merge.

[HaHaJeff]

Thanks. The ordering scan was originally added to validate custom store output before MOR. However, sorted output is already part of the RealtimeStore reader contract, and validating every adjacent row duplicated key projection and comparison performed by the downstream merge reader. Commit b2827df removes that hot-path validation and leaves sorting responsibility at the framework/store contract boundary.

[wangyong9999]

Per-row bitmap bookkeeping under a lock on the commit write path, to confirm the store returned each sealed offset exactly once. For the built-in store this is checking RawBatchReader's own output, and the guarantee is weak anyway: FinishReader only reports a gap once every reader reaches EOF, so an aborted or partially drained merge reports nothing.

Consider keeping only the cheap "offset inside the sealed range" bound on the hot path and moving the exactly-once bitmap behind a debug build.

[HaHaJeff]

The earlier count/min/max implementation was superseded by the exact commit-coverage requirement. At current PR head a7bc03b7b23c0302ad77d0e7312c0fb4da8a334c, commit readers share a RealtimeOffsetCoverage backed by RoaringBitmap64: each offset is checked for range and duplication, and complete sealed-range cardinality is verified after all readers reach EOF. Query readers do not use this coverage object. This keeps runtime validation because PK-sorted readers do not preserve offset monotonicity, so count/min/max cannot detect a duplicate-plus-hole case. Integration tests cover missing, duplicate, and out-of-range offsets.

[wangyong9999]

Each output batch copies its rows three times: one Take per contributing source, then Concatenate, then a second Take to restore merge order. The store keeps one StoredBatch per Write call and CreateQueryReaders merges every batch in the view, so a CDC-style writer with many small batches lands in this multi-source path nearly every time.

Take accepts a ChunkedArray with indices spanning chunks, so keeping the sources as one ChunkedArray and taking chunk_base[source] + row in a single call yields the same output with one copy and less code.

[HaHaJeff]

The store-side heap merge was originally introduced as a performance optimization: it bounded the number of readers returned for many small stored batches and reduced downstream merge inputs. However, it also required repeated Take/Concatenate operations and coupled the store to PK ordering and merge behavior. Commit ff44419 removes that duplicate merge layer and returns one zero-copy reader per prepared sorted batch. This trades higher reader cardinality for fewer copies and a storage-only plugin boundary. If reader cardinality becomes a measured bottleneck, sorted-run composition should be optimized in the framework instead.

[wangyong9999]

ValidatePrimaryKeyRealtimeOptions rejects every other unsupported option up front, but compaction options are silently dropped here instead: commit.force-compact becomes a no-op (NoopCompactManager::GetCompactionResult returns empty even when blocking), num-sorted-run.stop-trigger stops applying backpressure (ShouldWaitFor* always false), and level-0 runs grow unbounded until an external compactor runs. The read side pays for that growth directly, since PK realtime folds every disk split of a bucket into a single merge.

Rejecting explicitly-set compaction options in the validator would make this behave like the other unsupported options instead of quietly ignoring user configuration.

[HaHaJeff]

Thanks for pointing this out. The behavior is intentional and matches append realtime. With a realtime context, both writer paths use NoopCompactManager, reject explicit Compact(), and do not perform writer-local compaction during PrepareCommit. Compaction-related options are therefore not rejected; external compaction is responsible for rewriting accumulated files. For consistency with append realtime, this PR keeps the PK options unrejected.

[lxy-9602]

Why does store need the PK field here? It seems this may be for preparing key_comparator_, but in earlier discussions we explicitly agreed to remove sorting from the plugin side.

[HaHaJeff]

Good point. The PK fields were originally passed to the store because it constructed a comparator and performed store-side sorting and merging. That no longer matches the storage-only plugin boundary. Commit ff44419 removes the PK fields and comparator dependency, and b2827df simplifies the remaining selection to RealtimeStoreMode.

[lxy-9602]

Does the store currently use partition or bucket internally? These values seem to belong to the framework-side store registry and committed-progress identity rather than the store plugin contract. I suggest changing GetOrCreateRealtimeStore to accept a RealtimeStoreCreateRequest and a separate RealtimePartitionBucket, and removing partition and bucket from RealtimeStoreCreateRequest.

[HaHaJeff]

Fixed in 13cba67b9ef802f9acebe6b0b997d1e44dc528bf: partition and bucket were removed from RealtimeStoreCreateRequest, and a separate RealtimePartitionBucket is now passed to GetOrCreateRealtimeStore.

[lxy-9602]

We may also need to add RealtimeOffset here.

[HaHaJeff]

Fixed in 13cba67b9ef802f9acebe6b0b997d1e44dc528bf: _REALTIME_OFFSET is now included in SpecialFields::IsSystemField, with test coverage.

[lxy-9602]

May change func to FlushSortedReaders or WriteSortedReadersToFiles.

[HaHaJeff]

Fixed in 13cba67b9ef802f9acebe6b0b997d1e44dc528bf: the method was renamed to WriteSortedReadersToFiles.

[lxy-9602]

Could some of the checks here be extracted into a shared func and reused with GetRealtimeContext for append tables? It seems like AdvanceCommittedProgress may also contain similar logic.

[HaHaJeff]

Fixed in 13cba67b9ef802f9acebe6b0b997d1e44dc528bf: append and primary-key writer creation now share RestoreRealtimeCommittedProgress.

[lxy-9602]

Do we really need to fetch all fields here? In theory, wouldn’t it be enough to populate something like MergeFileSplitRead with the PK fields plus the sequence field?

[HaHaJeff]

Fixed in 51671a68ef295fd55613c1da974279d5f13383d4: the query schema is built from the MergeFileSplitRead value schema instead of the full table schema. The winning in-memory row still needs the projected output fields, not only the primary key and sequence fields.

[lxy-9602]

Why was this changed from directly using PAIMON_ASSIGN_OR_RAISE(...) to first storing the result in Result<std::shared_ptr<RealtimeStore>> store_result and then unwrapping it?

[HaHaJeff]

Fixed in 51671a68ef295fd55613c1da974279d5f13383d4: the direct PAIMON_ASSIGN_OR_RAISE usage was restored.

[lxy-9602]

PAIMON_RETURN_NOT_OK(func)could be enough here; PAIMON_RETURN_NOT_OK(func.status()) isn’t necessary.

[HaHaJeff]

Fixed in 51671a68ef295fd55613c1da974279d5f13383d4: the Result is passed directly to PAIMON_RETURN_NOT_OK, without calling .status().

[lxy-9602]

materialized_max_sequence_number seems to be used only to support writer handoff with uncommitted state retained in the same RealtimeContext. For failure recovery, the old context should be discarded, and the sequence number should be restored from the latest snapshot before replaying the data. Reusing this value would instead assign larger sequence numbers during replay. Could you confirm whether handoff with uncommitted state is a required use case? If not, I suggest removing this state and always restoring the sequence number from files.

[HaHaJeff]

The materialized watermark is not for failure recovery: failure discards the context and restores from the snapshot. It supports successful sequential handoff, matching append realtime context/store reuse; primary-key mode must additionally continue the synthetic sequence. I personally prefer one RealtimeContext per writer with no handoff because it simplifies the design, but the current code retains state to match the existing append lifecycle.

[lxy-9602]

This is a very elegant design 👍. We may be able to apply the same idea later to manage the lifetime of ArrowArrays returned by BatchReader, so that BatchReader no longer has to outlive the ArrowArray.

[HaHaJeff]

Thank you; I agree that this lifetime pattern may also be worth considering for BatchReader-returned Arrow arrays in a follow-up.

[zjw1111]

Thanks for the extensive refactor. Besides the inline comments, could you also clean up the remaining style/reuse items before merge?

• realtime_context_impl.cpp still uses string concatenation with std::to_string instead of fmt::format.
• primary_key_realtime_store.cpp and prepared_key_value_reader.cpp contain raw new outside the documented private-constructor factory exception.
• realtime_primary_key_writer.h uses std::map in its public signature without directly including <map>.

Could you also update the PR description to match the current implementation? It still mentions AppendRealtimeStoreCreateConfig, PrimaryKeyRealtimeStoreCreateConfig, and RealtimeStoreCreateConfig, which no longer exist, and still claims heap-based merging with constant query-reader cardinality even though the store now returns one reader per prepared batch.

[zjw1111]

Thanks for refining the projected memory schema. Could we include the complete trimmed primary key here in addition to the projected value fields? value_schema only contains primary-key fields explicitly requested by the user, while PreparedKeyValueReader resolves every field from the full key_schema. Therefore a query such as SELECT payload FROM pk_table fails with cannot find field id ... whenever a realtime memory view is present. Please construct a field-ID-deduplicated union of the full trimmed PK and projected value fields, while keeping the final output projection unchanged, and add single/composite-PK tests that omit key columns.

[HaHaJeff]

Fixed in 178ca71; reader setup was then simplified in 3c5c8e6. The read schema now uses a field-ID-deduplicated union of the full trimmed primary key and projected value fields. Covered by TestPkKeylessProjection and TestCompositePkKeylessProjection.

[zjw1111]

Could we bound the realtime merge fan-in here? RealtimeTableScan groups the entire partition-bucket into one realtime split, and this loop flattens every section's sorted runs before combining them with all memory readers in one sort-merge reader. The loser tree advances every run during initialization, so the number of leaves and retained first batches grows with accumulated disk runs and memory batches; writer-local compaction is disabled on this path. One option is to merge each disk section first, concatenate the non-overlapping section readers into one disk run, and only then merge that run with memory readers. A documented hard fan-in limit would also prevent unbounded resource use.

[HaHaJeff]

The disk side is fixed in 08a695c by composing the disk sections into one sorted disk reader before the final merge. The memory side is not fully fixed: fan-in remains unbounded by prepared-batch count, so many small writes increase final merge fan-in and retained first batches. This is a current performance/resource limitation rather than a correctness issue, and I am keeping this discussion open.

[zjw1111]

Could we move Levels::Create into the non-realtime branch? The realtime branch installs NoopCompactManager, and levels is only consumed by CreateCompactManager in the else branch. Restoring a realtime writer currently pays the traversal, grouping, set construction, and validation cost for all restored files even though the result is discarded.

[HaHaJeff]

Fixed in 4781383 by moving Levels::Create into the non-realtime compact-manager branch.

[zjw1111]

Could we reclaim the covered prefix with a single range erase (or use a deque)? Repeated erase(begin()) shifts the remaining vector on every iteration, making a refresh that reclaims many sealed segments O(n^2).

[HaHaJeff]

Fixed in 4781383 using one range erase, with coverage for multiple reclaimed segments and a pinned read view.

[zjw1111]

Could we centralize construction of the prepared transport schema in src/paimon/common/table/special_fields.h ? The same special-field prefix and nullability contract is currently assembled here, in RealtimePrimaryKeyWriter, and in KeyValueTableRead, while PreparedKeyValueReader separately hard-codes the corresponding indexes. This is part of the public plugin protocol, so a single helper would prevent the write, commit, and query paths from drifting.

[HaHaJeff]

Fixed in 178ca71 and simplified in 3c5c8e6. SpecialFields::PreparedKeyValueSchema and shared indexes now own the prepared transport protocol, with focused test coverage.

[zjw1111]

Is PK realtime v1 intended to identify a real versioned API or serialized-format contract? I could not find a corresponding version constant or version dispatch, and the PR explicitly introduces no new data-file format. If v1 is only a temporary implementation-phase label, could we remove it from user-facing errors (for example, use PK realtime or the current PK realtime implementation) so it does not imply persistence or protocol compatibility semantics? If it is intentional versioning, please document what is versioned and where compatibility is enforced.

[HaHaJeff]

Fixed in 4781383. v1 was not a protocol or API version, so it was removed from user-facing errors.

[zjw1111]

Additionally, you could also review the code to identify any unnecessary validations or overly defensive code generated by AI, and remove some of it where appropriate.

[HaHaJeff]

Follow-up for review PRR_kwDOSj74F88AAAABK7352w and issue comment 5423886006: the style fixes are in 4781383, and the remaining raw new calls are only private-constructor Create factory exceptions. The PR description is now updated. Cleanup commit 824389e removes the duplicate private visible-offset check; real plugin/C Data boundary validations remain intentionally.

[lxy-9602]

I understand this was part of the refactor, but could we keep the original comments instead of removing them?

[HaHaJeff]

Restored the original comment above the corresponding KeyValueProjectionReader construction in 982eb59256ebf14d943e202950b35ad30b8fc671.

[lxy-9602]

Could we keep the tests in this PR focused on the basic streaming PK functionality and core scenarios, and move exception-related cases into a separate PR? The current fault-injection style is a bit different from the rest of the codebase, so reviewing it separately would make it easier to understand and evaluate. Also, for some classes like SplitBatchReader, it doesn’t seem like the current code path would actually hit the simulated issues, so it might make more sense to add that kind of simulation when the scenario really arises.

[HaHaJeff]

Agreed, and I narrowed the test scope in 813c6444e39dba4076018a34126a3e3b3dfc176c. The custom fault-injection helpers, SplitBatchReader, and the related exception and malformed-reader cases were removed, leaving the basic streaming PK and core lifecycle coverage in this PR. Those exception paths are not covered here anymore; they can be added separately once we have representative production scenarios for them.

[zjw1111]

Thanks for streamlining the realtime test coverage. Could you make two follow-up cleanups before merge?

1. Please centralize the remaining PK realtime transport-schema construction in tests through RealtimePrimaryKeyLayout::CreateSchema. The production write/read paths already use this helper, but the special-field prefix is still assembled manually in:

   • src/paimon/core/operation/key_value_file_store_write_test.cpp
   • src/paimon/core/realtime/primary_key_realtime_store_test.cpp (TransportSchema, NestedTransportSchema, and the nested-projection schema)
   • test/inte/realtime_write_inte_test.cpp (ReadPkSequences)

   These test-local copies can drift from RealtimePrimaryKeyLayout. Please construct only the value fields at those sites and let CreateSchema add _VALUE_KIND, _SEQUENCE_NUMBER, and _REALTIME_OFFSET. Tests that intentionally mutate a helper-created schema to exercise ValidateSchema should remain as they are.

2. Please synchronize the PR description with the current head:

   • It still says that SpecialFields::PreparedKeyValueSchema and shared prepared-field indexes define the transport schema, but those symbols no longer exist; the current API is RealtimePrimaryKeyLayout::CreateSchema / ValidateSchema with the layout indexes.
   • It says that integration tests cover prepare-commit, commit, write, and refresh failure recovery with context/writer recreation and external WAL replay. Commit 813c6444 removed the fault-injection helpers and the related recovery integration tests, so this coverage claim, and any test totals affected by that cleanup, should be updated. The failure-recovery behavior may remain documented as a caller contract if intended, but it should not be presented as integration coverage that is still in this PR.

[lxy-9602]

Could this behavior be covered in a focused reader adapter unit test instead? Verifying that Close() propagates to the underlying reader seems valuable, but testing it here requires several store/factory wrappers and adds considerable integration-test scaffolding.

[zjw1111]

In my view, this test can be removed entirely. The components in this reader chain and most of their Close() propagation logic already existed in the framework before this PR, so they should not be tested specifically as part of the realtime PK implementation.

[zjw1111]

One test-scaffolding cleanup suggestion.

[zjw1111]

Thanks for adding the lifetime coverage here. Could you evaluate whether two test-only wrappers can be removed without reducing that coverage?

• This TestingMemoryPool only forwards to GetMemoryPool() and is used to observe lifetime through a weak_ptr. It seems the test could instead use std::shared_ptr<MemoryPool> pool = GetMemoryPool() together with std::weak_ptr<MemoryPool>.
• ReadViewCheckingBatchReader in test/inte/realtime_write_inte_test.cpp checks the same read-view lifetime on NextBatch(), while TestPkRead already asserts that the tracked view remains alive after the context is destroyed and immediately before reading, then expires after the reader is closed.

Would it be possible to remove these two wrappers while keeping the existing explicit lifetime assertions? I am only suggesting a test-scaffolding cleanup; the pool and read-view lifetime coverage itself should remain.

[HaHaJeff]

Both test-only wrappers were removed in commit a77b5f4. The pool test now uses shared_ptr and weak_ptr directly, while TestPkRead retains the explicit read-view lifetime assertions before reading and after closing the reader.

[zjw1111]

Please remove the id number or use the correct id.

[HaHaJeff]

Follow-up to issue comment 5450741187: Both follow-ups are addressed. The remaining test-local transport schemas now use RealtimePrimaryKeyLayout::CreateSchema in a77b5f4; the tests provide only the value fields, while the layout helper supplies the transport fields. The intentional malformed-schema validation cases remain unchanged. I also synchronized the PR description with the current implementation and test scope. It now refers to RealtimePrimaryKeyLayout::CreateSchema / ValidateSchema, removes the obsolete fault-injection recovery coverage claim, keeps failure recovery as a caller contract, and reports the current totals of 39 focused core tests and 57 realtime integration tests.