From 1e08ef8dc72bfe079e2f14f2d85a888287bd7a4e Mon Sep 17 00:00:00 2001 From: deardeng Date: Tue, 28 Jul 2026 18:41:56 +0800 Subject: [PATCH] [fix](load) correct quorum participants for incremental streams (#66016) pick from https://github.com/apache/doris/pull/66016 Exclude streams that have not entered the current close stage from V2 quorum accounting. Incremental streams now require CLOSE_LOAD participation before contributing to quorum success, preserving the two-stage close fence and preventing premature tablet commits. --- be/src/vec/sink/load_stream_stub.h | 2 ++ be/src/vec/sink/writer/vtablet_writer_v2.cpp | 24 ++++++++++++++++---- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/be/src/vec/sink/load_stream_stub.h b/be/src/vec/sink/load_stream_stub.h index ad1c42b42c7443..7f240c3c81dedd 100644 --- a/be/src/vec/sink/load_stream_stub.h +++ b/be/src/vec/sink/load_stream_stub.h @@ -217,6 +217,8 @@ class LoadStreamStub : public std::enable_shared_from_this { bool is_open() const { return _is_open.load(); } + bool is_closing() const { return _is_closing.load(); } + bool is_incremental() const { return _is_incremental; } friend std::ostream& operator<<(std::ostream& ostr, const LoadStreamStub& stub); diff --git a/be/src/vec/sink/writer/vtablet_writer_v2.cpp b/be/src/vec/sink/writer/vtablet_writer_v2.cpp index b3e306357e2c48..0a7eada876a770 100644 --- a/be/src/vec/sink/writer/vtablet_writer_v2.cpp +++ b/be/src/vec/sink/writer/vtablet_writer_v2.cpp @@ -718,9 +718,23 @@ Status VTabletWriterV2::close(Status exec_status) { // close_wait on all non-incremental streams, even if this is not the last sink. // because some per-instance data structures are now shared among all sinks // due to sharing delta writers and load stream stubs. - // Do not need to wait after quorum success, - // for first-stage close_wait only ensure incremental streams load has been completed, - // unified waiting in the second-stage close_wait. + // + // This stage is also a cross-source fence for a source that has incremental streams: + // it must not close those streams before every other source has entered the close + // phase and can no longer open new incremental streams. + // + // A stream contributes to quorum only after CLOSE_LOAD has been sent + // (LoadStreamStub::is_closing) and the sender has observed both EOS and StreamClose. + // When this source has incremental streams, its non-incremental CLOSE_LOAD carries + // num_incremental_streams > 0, so the destination defers StreamClose until CLOSE_LOAD + // has arrived from all sources (LoadStream::_dispatch). Therefore, any non-incremental + // stream counted towards quorum is a valid lifecycle fence for this source. + // + // A source without incremental streams does not require this fence because it has no + // incremental streams to close. + // + // The remaining streams do not need to be waited for in this stage; they are included + // in the second-stage close_wait. RETURN_IF_ERROR(_close_wait(_non_incremental_streams(), false)); // send CLOSE_LOAD on all incremental streams if this is the last sink. @@ -890,7 +904,9 @@ bool VTabletWriterV2::_quorum_success( for (const auto& [dst_id, streams] : streams_for_node) { bool finished = true; for (const auto& stream : streams->streams()) { - if (unfinished_streams.contains(stream) || !stream->check_cancel().ok()) { + // Incremental streams do not participate in the first close stage. + if (!stream->is_closing() || unfinished_streams.contains(stream) || + !stream->check_cancel().ok()) { finished = false; break; }