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