fix(realtime): keep spill outside the write mutex

include/paimon/realtime/realtime_store.h

@@ -145,7 +145,9 @@ struct PAIMON_EXPORT RealtimeQueryContext {
 /// Customizable plugin interface for storing and querying real-time rows before Paimon data-file
 /// generation.
 ///
-/// Paimon serializes calls to `Write` and `SealForCommit` for the same store. After sealing,
+/// Calls to `Write` and `SealForCommit` can overlap for the same store. Store implementations must
+/// synchronize segment rotation. After rotation, subsequent writes go to the new building segment
+/// while the sealed segment may be spilled. After sealing,
 /// `CreateCommitReaders` may read the immutable sealed segment while later `Write` calls append to
 /// a new building segment. Paimon retains control of file format, rolling, indexes, and
 /// commit-message generation. A store may choose its own in-memory representation, indexes, and

src/paimon/core/realtime/realtime_append_only_writer.cpp

@@ -121,27 +121,29 @@ Status RealtimeAppendOnlyWriter::Write(std::unique_ptr<RecordBatch>&& batch) {
 Status RealtimeAppendOnlyWriter::SealCurrentSegment() {
     PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<RealtimeSegmentHandle>> segment,
                            realtime_store_->SealForCommit());
-    if (segment) {
-        if (!segment.value()) {
-            return Status::Invalid("append real-time store sealed a null segment");
-        }
-        sealed_segments_.push_back(std::move(segment.value()));
-        has_building_data_ = false;
+    std::lock_guard<std::mutex> lock(realtime_store_mutex_);
+    if (!segment) {
+        return Status::OK();
     }
+    if (!segment.value()) {
+        return Status::Invalid("append real-time store sealed a null segment");
+    }
+    has_building_data_ = next_offset_ > segment.value()->GetOffsetRange().end;
+    sealed_segments_.push_back(std::move(segment.value()));
     return Status::OK();
 }
 
 Status RealtimeAppendOnlyWriter::Seal() {
-    std::lock_guard<std::mutex> lock(realtime_store_mutex_);
+    std::lock_guard<std::mutex> lock(prepare_mutex_);
     return SealCurrentSegment();
 }
 
 Result<CommitIncrement> RealtimeAppendOnlyWriter::PrepareCommit(bool wait_compaction) {
     std::lock_guard<std::mutex> lock(prepare_mutex_);
+    PAIMON_RETURN_NOT_OK(SealCurrentSegment());
     std::vector<std::shared_ptr<RealtimeSegmentHandle>> segments;
     {
         std::lock_guard<std::mutex> realtime_store_lock(realtime_store_mutex_);
-        PAIMON_RETURN_NOT_OK(SealCurrentSegment());
         segments.swap(sealed_segments_);
     }
     for (const std::shared_ptr<RealtimeSegmentHandle>& segment : segments) {

src/paimon/core/realtime/realtime_primary_key_writer.cpp

@@ -209,27 +209,29 @@ Status RealtimePrimaryKeyWriter::Write(std::unique_ptr<RecordBatch>&& batch) {
 Status RealtimePrimaryKeyWriter::SealCurrentSegment() {
     PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<RealtimeSegmentHandle>> segment,
                            realtime_store_->SealForCommit());
-    if (segment) {
-        if (!segment.value()) {
-            return Status::Invalid("PK real-time store sealed a null segment");
-        }
-        sealed_segments_.push_back(std::move(segment.value()));
-        has_building_data_ = false;
+    std::lock_guard<std::mutex> lock(realtime_store_mutex_);
+    if (!segment) {
+        return Status::OK();
     }
+    if (!segment.value()) {
+        return Status::Invalid("PK real-time store sealed a null segment");
+    }
+    has_building_data_ = next_offset_ > segment.value()->GetOffsetRange().end;
+    sealed_segments_.push_back(std::move(segment.value()));
     return Status::OK();
 }
 
 Status RealtimePrimaryKeyWriter::Seal() {
-    std::lock_guard<std::mutex> lock(realtime_store_mutex_);
+    std::lock_guard<std::mutex> lock(prepare_mutex_);
     return SealCurrentSegment();
 }
 
 Result<CommitIncrement> RealtimePrimaryKeyWriter::PrepareCommit(bool wait_compaction) {
     std::lock_guard<std::mutex> prepare_lock(prepare_mutex_);
+    PAIMON_RETURN_NOT_OK(SealCurrentSegment());
     std::vector<std::shared_ptr<RealtimeSegmentHandle>> segments;
     {
         std::lock_guard<std::mutex> store_lock(realtime_store_mutex_);
-        PAIMON_RETURN_NOT_OK(SealCurrentSegment());
         segments.swap(sealed_segments_);
     }
     for (const std::shared_ptr<RealtimeSegmentHandle>& segment : segments) {