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