HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3835876940
##########
include/paimon/realtime/realtime_store.h:
##########
@@ -40,6 +42,29 @@ namespace paimon {
class MemoryPool;
class Predicate;
+struct PAIMON_EXPORT AppendRealtimeStoreCreateConfig {};
+
+struct PAIMON_EXPORT PrimaryKeyRealtimeStoreCreateConfig {
+ std::vector<std::string> primary_keys;
+ /// Largest sequence restored from the committed snapshot. A PK store
assigns one contiguous
+ /// sequence to every mutation in `Write` order, starting at the next
value, and rejects
+ /// `Write` before the assigned sequence would exceed `INT64_MAX - 1`.
+ int64_t restore_max_sequence_number;
+};
+
+using RealtimeStoreCreateConfig =
+ std::variant<AppendRealtimeStoreCreateConfig,
PrimaryKeyRealtimeStoreCreateConfig>;
+
+struct PAIMON_EXPORT RealtimeStoreCreateRequest {
+ /// Complete table write schema whose ownership is transferred to the
factory.
+ std::unique_ptr<::ArrowSchema> write_schema;
+ std::map<std::string, std::string> options;
+ std::shared_ptr<MemoryPool> memory_pool;
+ std::map<std::string, std::string> partition;
+ int32_t bucket = -1;
+ RealtimeStoreCreateConfig mode_config;
+};
+
Review Comment:
> 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
`KeyValueRecordReader`s 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 `RecordBatch`es 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
`KeyValueRecordReader`s;
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 `BatchReader`s into sorted `KeyValueRecordReader`s;
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:
```text
RealtimeStore
-> BatchReader
-> framework BatchReader-to-KeyValueRecordReader adapter
-> MergeTreeWriter::WriteSortedReaders
-> existing SortMergeReader and merge functions
-> existing rolling data-file writer
-> CommitIncrement
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
