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

[lxy-9602]

Purpose

Linked issue: #158
This PR introduces an opt-in real-time write and read framework for fixed-bucket append tables.

Applications can attach a shared RealtimeContext to write and scan contexts. Writes are buffered by
a pluggable MemIndexer and become queryable before snapshot commit. During prepare commit, the
current memory segment is sealed and flushed through Paimon's existing rolling writer, while later
writes continue in a new segment.

The read path captures an immutable memory view and combines it with the latest committed disk
snapshot through RealtimeSplit. The append reader concatenates disk and memory readers, using
per-partition-bucket _OFFSET progress to avoid duplicate or missing rows.

The main changes are:

• Add the pluggable MemIndexer and MemIndexerFactory interfaces.
• Provide a default Arrow-backed memory indexer.
• Add RealtimeContext for sharing memory indexers between writers and readers.
• Add PrepareCommitWithProgress and CommitWithProgress.
• Seal memory segments during prepare commit and flush them through existing Paimon file writers.
• Persist per-partition-bucket committed offsets in versioned metadata/<uuid>.offsets files.
• Restore the next write offset from the latest committed snapshot.
• Add real-time table scans and disk-memory union reads.
• Capture immutable memory read views so an existing query remains stable across writes and refresh.
• Add RefreshCommittedSnapshot to reclaim sealed memory covered by committed disk data.
• Support projection, optional exact predicate filtering, partition filters, and bucket filters.
• 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:

• real-time offset JSON serialization and parsing;
• offset-file read, write, and snapshot-property merging;
• unordered commit progress sorting;
• contiguous offset validation and invalid range handling;
• real-time writer and commit API validation;
• writer memory-manager behavior.

Added 14 integration tests covering:

• append write, commit, and disk read;
• memory-only reads before prepare commit;
• disk and building-memory union reads;
• rolling files with continuous _OFFSET values;
• projection and exact predicate filtering;
• Parquet disk predicate pushdown without memory filtering;
• committed snapshot refresh and memory reclamation;
• immutable plans across refresh;
• repeated write, commit, read, and refresh cycles;
• concurrent write, prepare commit, read, commit, and refresh;
• multiple partitions and multiple buckets;
• independent bucket-offset recovery;
• offset restoration from a committed snapshot.

API and Format

This PR adds the following public API concepts:

• RealtimeContext
• MemIndexer, MemIndexerFactory, and MemReadView
• RealtimeWriteBatch and RealtimeCommitProgress
• WriteContextBuilder::WithRealtimeContext
• ScanContextBuilder::WithRealtimeContext
• FileStoreWrite::PrepareCommitWithProgress
• FileStoreWrite::RefreshCommittedSnapshot
• FileStoreCommit::CommitWithProgress

The feature is opt-in through RealtimeContext; existing write, commit, and disk-only read paths are
unchanged when no real-time context is supplied.

Real-time data files contain the reserved _OFFSET field. Each committed snapshot references a
versioned offsets file containing the latest offset for every partition-bucket. Data files and
offset progress are published atomically by the same snapshot commit.

Real-time commits currently fail directly on snapshot conflicts. Failure recovery and idempotent
retry will be handled separately.

Documentation

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.