Skip to content

[fix](load) correct quorum participants for incremental streams (#66016) #67215

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deardeng wants to merge 1 commit into
base: branch-4.0
Choose a base branch
from
+22 −4
Open

[fix](load) correct quorum participants for incremental streams (#66016) #67215

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions be/src/vec/sink/load_stream_stub.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,8 @@ class LoadStreamStub : public std::enable_shared_from_this<LoadStreamStub> {

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);
Expand Down
24 changes: 20 additions & 4 deletions be/src/vec/sink/writer/vtablet_writer_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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;
}
Expand Down
Loading