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 creates an append or primary-key store from a typed request. The primary-key store keeps mutations in memory and passes sealed readers to MergeTreeWriter during prepare commit.

For primary-key reads, immutable memory read views are combined with the selected disk snapshot through RealtimeSplit. Both sources enter the existing primary-key merge-on-read path, including sequence ordering, row-kind handling, and merge-engine semantics. Predicates that are unsafe before merge are evaluated after merge.

The main changes are:

• add an in-memory primary-key RealtimeStore implementation;
• extend RealtimeStoreFactory with typed append and primary-key creation requests;
• let MergeTreeWriter consume sealed readers produced 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 built-in V1 primary-key implementation supports fixed-bucket tables with the deduplicate merge engine, full-row mutations, latest-snapshot recovery, concurrent readers, and internally synchronized write and prepare operations. Dynamic buckets, lookup or early MOR, aggregation and partial-update merge engines, data evolution, user sequence fields, explicit realtime-writer compaction, and recovery from a non-latest snapshot are not included.

The built-in V1 primary-key store keeps realtime mutations entirely in memory and does not implement spill. The public RealtimeStore contract still permits custom implementations to use their own spill strategy.

Tests

Added unit coverage for:

• typed factory creation for append and primary-key stores;
• primary-key write validation, sealing, and foreign-handle rejection;
• sequence ordering, row kinds, deduplication, and commit readers;
• immutable read views and concurrent readers;
• realtime offset progress and snapshot refresh; and
• supported option validation and no-spill writer construction.

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;
• deletes, repeated keys, projection, predicates, external compaction, and writer handoff;
• concurrent write, prepare, commit, refresh, and read operations; and
• non-realtime and append-realtime regression paths.

The focused primary-key realtime tests pass under ASAN, UBSAN, and LeakSanitizer.

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.

The factory API gains typed creation data:

• AppendRealtimeStoreCreateConfig for append tables;
• PrimaryKeyRealtimeStoreCreateConfig for primary-key tables;
• RealtimeStoreCreateConfig as the request variant; and
• RealtimeStoreCreateRequest for schema, options, memory pool, partition-bucket identity, and table-specific configuration.

Custom factories implement RealtimeStoreFactory::Create(RealtimeStoreCreateRequest&&) and dispatch on the append or primary-key configuration. A primary-key implementation returns primary-key-sorted mutations with the row kind and sequence metadata required by the existing merge-on-read path.

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 progress identifiers assigned by the framework and are 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 or prepare failure, the caller discards the writer and context, recreates them from the latest committed snapshot, and replays its external WAL.

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

Documentation

The public headers document typed store creation, the primary-key reader contract, offset and sequence separation, supported V1 options, lifecycle conventions, and ownership of returned Arrow data.

Generative AI tooling

Generated-by: Codex (GPT-5)

Reviewed-by: Claude Code (Claude Opus 4.8)

[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.

[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.

[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.