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 written transport batches and returns one reader per stored write batch. Query views include both sealed and currently building batches. 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. Reusing a store with a different transport schema is rejected, and realtime data evolution is not supported.

For each primary-key partition-bucket with active memory, all selected disk splits are folded into one RealtimeSplit; disk-only partition-buckets retain their ordinary disk splits. 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 retained memory write batch 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 the primary-key framework adapt store readers into sorted key-value readers that MergeTreeWriter can write;
• 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. Sequential writer handoff is supported after closing the prior writer. Reusing the same RealtimeContext continues from retained in-memory progress, while recreating the context restores offsets and live-file sequence progress from the latest 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. Building and sealed batches remain retained until committed-offset refresh reclaims them, and immutable read views may keep reclaimed segments alive. write-buffer-size does not bound this usage. 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 transport-schema checks;
• sequence ordering, row kinds, deduplication, and commit readers;
• one reader per stored write 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.

Validated with 39 focused core tests and all 57 realtime integration tests.

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 primary-key transport schema, options, memory pool, statistics mode, and RealtimeStoreMode; factories dispatch on the mode. For primary-key mode, the framework supplies batches with the primary-key transport schema and assigns offsets and sequence numbers before calling the store. statistics_mode configures the built-in append-only store; the built-in primary-key store does not collect or use in-memory min/max statistics.

The store returns raw transport-batch readers. For primary-key queries, offset_begin is ignored by the store. The store applies the requested transport projection, while the framework applies offset visibility, merge-on-read, predicates, and the final user projection. 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.

CommitWithProgress failures may be retried with the same arguments as documented by its public API. After a write or prepare-commit failure whose in-memory effects are unknown, the caller must discard the writer and realtime context, recreate them from the latest committed snapshot, and replay its external WAL. Refresh failures caused by removed or backward committed progress likewise require recreation; incomplete reclamation may be retried with the same snapshot. 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, the primary-key transport schema, 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.

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

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

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]

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.

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

[zjw1111]

One test-scaffolding cleanup suggestion.

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