HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3835876255
##########
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:
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]
