HaHaJeff commented on code in PR #163:
URL: https://github.com/apache/paimon-cpp/pull/163#discussion_r3703577646
##########
src/paimon/core/operation/abstract_file_store_write.cpp:
##########
@@ -261,7 +266,74 @@ Result<std::vector<std::shared_ptr<CommitMessage>>>
AbstractFileStoreWrite::Prep
return result;
}
+Result<std::vector<RealtimeCommitProgress>>
AbstractFileStoreWrite::PrepareCommitWithProgress(
+ int64_t) {
+ if (!IsRealtimeWrite()) {
+ return Status::Invalid("PrepareCommitWithProgress is only supported by
a real-time writer");
+ }
+ if (is_streaming_mode_ == false) {
+ return Status::Invalid("PrepareCommitWithProgress requires streaming
mode");
+ }
+ // Real-time prepare snapshots writers under a short lock so writes can
continue while sealed
+ // segments are flushed. The normal path iterates and may erase writers in
place, which would
+ // either race with concurrent writer creation or hold writers_mutex_ for
the whole prepare.
+ return PrepareRealtimeCommit();
+}
+
+Result<std::vector<RealtimeCommitProgress>>
AbstractFileStoreWrite::PrepareRealtimeCommit() {
+ struct WriterSnapshot {
+ BinaryRow partition;
+ int32_t bucket;
+ int32_t total_buckets;
+ std::shared_ptr<BatchWriter> writer;
+ };
+ std::vector<WriterSnapshot> writer_snapshots;
+ {
+ std::lock_guard<std::mutex> lock(writers_mutex_);
+ for (const auto& [partition, buckets] : writers_) {
+ for (const auto& [bucket, container] : buckets) {
+ writer_snapshots.push_back(
+ WriterSnapshot{partition, bucket, container.total_buckets,
container.writer});
+ }
+ }
+ }
+
+ std::vector<RealtimeCommitProgress> result;
+ for (const WriterSnapshot& snapshot : writer_snapshots) {
Review Comment:
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.
--
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]
