This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 6fa3c815 fix(realtime): keep spill outside the write mutex (#384)
6fa3c815 is described below
commit 6fa3c815937db504dcc477bdb7f8696182e4300e
Author: JeffZhou <[email protected]>
AuthorDate: Wed Sep 23 21:05:39 2026 +0800
fix(realtime): keep spill outside the write mutex (#384)
---
include/paimon/realtime/realtime_store.h | 4 +++-
.../core/realtime/realtime_append_only_writer.cpp | 18 ++++++++++--------
.../core/realtime/realtime_primary_key_writer.cpp | 18 ++++++++++--------
3 files changed, 23 insertions(+), 17 deletions(-)
diff --git a/include/paimon/realtime/realtime_store.h
b/include/paimon/realtime/realtime_store.h
index c15bc26a..1d9ab0c6 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 6b89fd51..a5764685 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<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) {
diff --git a/src/paimon/core/realtime/realtime_primary_key_writer.cpp
b/src/paimon/core/realtime/realtime_primary_key_writer.cpp
index 1afe8781..05d534a1 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<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) {
