diff --git a/include/paimon/realtime/realtime_store.h b/include/paimon/realtime/realtime_store.h index c15bc26ae..1d9ab0c66 100644 --- a/include/paimon/realtime/realtime_store.h +++ b/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 diff --git a/src/paimon/core/realtime/realtime_append_only_writer.cpp b/src/paimon/core/realtime/realtime_append_only_writer.cpp index 6b89fd51d..a57646853 100644 --- a/src/paimon/core/realtime/realtime_append_only_writer.cpp +++ b/src/paimon/core/realtime/realtime_append_only_writer.cpp @@ -121,27 +121,29 @@ Status RealtimeAppendOnlyWriter::Write(std::unique_ptr&& batch) { Status RealtimeAppendOnlyWriter::SealCurrentSegment() { PAIMON_ASSIGN_OR_RAISE(std::optional> 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 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 lock(realtime_store_mutex_); + std::lock_guard lock(prepare_mutex_); return SealCurrentSegment(); } Result RealtimeAppendOnlyWriter::PrepareCommit(bool wait_compaction) { std::lock_guard lock(prepare_mutex_); + PAIMON_RETURN_NOT_OK(SealCurrentSegment()); std::vector> segments; { std::lock_guard realtime_store_lock(realtime_store_mutex_); - PAIMON_RETURN_NOT_OK(SealCurrentSegment()); segments.swap(sealed_segments_); } for (const std::shared_ptr& segment : segments) { diff --git a/src/paimon/core/realtime/realtime_primary_key_writer.cpp b/src/paimon/core/realtime/realtime_primary_key_writer.cpp index 1afe8781b..05d534a1a 100644 --- a/src/paimon/core/realtime/realtime_primary_key_writer.cpp +++ b/src/paimon/core/realtime/realtime_primary_key_writer.cpp @@ -209,27 +209,29 @@ Status RealtimePrimaryKeyWriter::Write(std::unique_ptr&& batch) { Status RealtimePrimaryKeyWriter::SealCurrentSegment() { PAIMON_ASSIGN_OR_RAISE(std::optional> 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 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 lock(realtime_store_mutex_); + std::lock_guard lock(prepare_mutex_); return SealCurrentSegment(); } Result RealtimePrimaryKeyWriter::PrepareCommit(bool wait_compaction) { std::lock_guard prepare_lock(prepare_mutex_); + PAIMON_RETURN_NOT_OK(SealCurrentSegment()); std::vector> segments; { std::lock_guard store_lock(realtime_store_mutex_); - PAIMON_RETURN_NOT_OK(SealCurrentSegment()); segments.swap(sealed_segments_); } for (const std::shared_ptr& segment : segments) {