feat: support real-time append writes and reads with pluggable memory indexers

[lxy-9602]

Purpose

Linked issue: #158

This PR introduces an opt-in, process-local real-time write and union-read framework for
fixed-bucket append tables.

Applications attach the same RealtimeContext to their write and scan contexts. Each
partition-bucket is backed by a pluggable MemIndexer, so newly written rows become queryable
before the next snapshot commit. During prepare commit, Paimon seals the current segment, opens a
new building segment for subsequent writes, and flushes the sealed segment through the existing
rolling writer. Paimon therefore continues to own data-file formats, file rolling, file indexes,
statistics, and commit-message generation.

The framework assigns an internal contiguous offset range to each written batch. These offsets are
segment and snapshot progress metadata. PrepareCommitWithProgress returns each commit message
together with its partition, bucket, and inclusive offset range. CommitWithProgress orders and
validates these ranges against the latest committed prefix, then atomically publishes the data files
and updated progress.

For reads, Paimon captures immutable MemReadView objects and combines them with the selected disk
snapshot through RealtimeSplit. The append reader concatenates disk readers with memory readers
whose offsets are newer than that snapshot's committed partition-bucket boundary. Refreshing a
committed snapshot advances the shared context and allows fully covered sealed segments to be
reclaimed, while existing query plans continue to pin their original memory views.

The main changes are:

• Add pluggable MemIndexer, MemIndexerFactory, RealtimeSegmentHandle, and
  MemReadView interfaces.
• Provide a default Arrow-backed memory indexer.
• Add RealtimeContext to own partition-bucket indexers and share them between writers and
  readers.
• Add PrepareCommitWithProgress and CommitWithProgress.
• Seal memory segments during prepare commit and flush them through existing Paimon file writers.
• Persist versioned per-partition-bucket committed progress in metadata/<uuid>.offsets files
  referenced by snapshot properties.
• Restore the next internal offset from the latest committed snapshot.
• Add real-time table scans, RealtimeSplit, and append disk-memory union reads.
• Capture immutable memory views so a query remains stable across concurrent writes and refresh.
• Reclaim sealed memory after committed progress advances.
• Support projection, partition and bucket filters, framework-level exact predicate filtering, and
  optional plugin predicate pushdown.
• Preserve existing rolling, file-format, file-index, statistics, and physical-field handling.

The current implementation supports streaming writes and latest-snapshot batch scans for
fixed-bucket append tables. Primary-key tables, deletion vectors, data evolution, streaming scans,
scan-limit pushdown, and global-index splits are not included in this PR.

Tests

Added unit coverage for:

• Arrow memory-indexer write, seal, query-boundary, reclamation, and close behavior;
• real-time context indexer reuse, immutable read views, offset restoration, and monotonic committed
  progress;
• versioned offset JSON serialization, validation, offset-file I/O, and snapshot-property merging;
• unordered commit-progress sorting and contiguous-prefix validation;
• real-time writer range tracking and API validation;
• predicate field binding and writer memory-manager behavior.

Added 14 integration tests covering:

• append write, commit, and disk read;
• rolling files while preserving continuous commit progress;
• ordering commit messages prepared out of order;
• memory-only reads before prepare commit;
• disk and building-memory union reads;
• projection and exact predicate filtering across disk and memory;
• Parquet disk predicate pushdown without memory filtering;
• committed snapshot refresh and memory reclamation;
• immutable query plans across refresh;
• repeated write, commit, read, and refresh cycles;
• concurrent write, prepare commit, read, commit, and refresh;
• disk-memory combinations across multiple partitions;
• independent progress restoration across multiple buckets;
• internal offset restoration from a committed snapshot.

API and Format

This PR adds the following public API concepts:

• RealtimeContext, RealtimePartitionBucket, and RealtimeOffsetMap;
• MemIndexer, MemIndexerFactory, RealtimeSegmentHandle, and MemReadView;
• RealtimeWriteBatch, MemQueryContext, and RealtimeCommitProgress;
• WriteContextBuilder::WithRealtimeContext;
• ScanContextBuilder::WithRealtimeContext;
• FileStoreWrite::PrepareCommitWithProgress;
• FileStoreWrite::RefreshCommittedSnapshot;
• FileStoreCommit::CommitWithProgress.

The feature is enabled only when a RealtimeContext is supplied. Existing write, commit, and
disk-only read paths remain unchanged otherwise.

This PR adds a versioned offset metadata file referenced by the realtime.offsets snapshot property. The file stores the largest committed internal offset for each logical partition and bucket; data files and progress metadata are published by the same snapshot commit.

Real-time commits currently fail directly on snapshot conflicts instead of retrying with stale
progress. Failure recovery and idempotent commit retry can be addressed separately.

Documentation

This is a new opt-in feature. The public interfaces contain API documentation, and the design is
tracked in #158.

Generative AI tooling

Generated-by: OpenAI Codex (GPT-5)

[HaHaJeff]

How should realtime writes recover if PrepareCommitWithProgress fails for
one bucket?

Terminating all bucket writers would make the failure scope too large for a
realtime workload. Could we preserve the prepared state of successful buckets
and retry or recreate only the failed bucket writer?

[lxy-9602]

Thanks for raising this concern. Bucket-level failure isolation is indeed valuable for long-running real-time workloads.

We checked the existing Paimon write path, including the Java Spark integration. Currently, FileStoreWrite.prepareCommit does not provide partial-success semantics across buckets. If preparing any bucket fails, the exception is propagated through TableWrite to the Spark data writer, causing the corresponding Spark task attempt to fail. Commit messages from buckets prepared earlier in the same call are not returned as independently recoverable results.

More generally, Paimon’s current read and write APIs do not define a contract for partial failure and partial recovery within one operation. For the first phase, we would prefer to keep the real-time implementation consistent with this existing behavior rather than introduce a separate recovery model only for PrepareCommitWithProgress.

For your use case, one practical approach is to use one FileStoreWrite instance per bucket. This keeps the failure scope local to that bucket: a failed bucket writer can be recreated and replayed independently, while writers for other buckets remain unaffected. A higher-level coordinator can still collect their commit messages and commit them under the desired snapshot boundary.

Bucket-level prepared-state preservation and retry could be considered as a future enhancement, but it would require a broader partial-recovery contract across Paimon’s read and write paths.

[HaHaJeff]

Thanks for the clarification. This approach works for us.

We can use one FileStoreWrite per partition/bucket, recover a failed writer
independently, and let a higher-level coordinator collect the progress and
commit it under the same snapshot boundary.