This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d90b22cb858 [fix](load) correct quorum participants for incremental
streams (#66016)
d90b22cb858 is described below
commit d90b22cb858e4db98b1e3e1dd3c14b2da16e2cc2
Author: deardeng <[email protected]>
AuthorDate: Tue Jul 28 18:41:56 2026 +0800
[fix](load) correct quorum participants for incremental streams (#66016)
### What problem does this PR solve?
Problem Summary:
Auto partition loads may create incremental load streams after another
source has already reached write quorum.
In `VTabletWriterV2::_quorum_success()`, a stream that had not entered
the current close stage was implicitly counted as finished — the check
only looked at `unfinished_streams` membership and cancellation.
Incremental streams never enter the first-stage `close_wait` participant
set (`_non_incremental_streams()`), so they contributed free "finished"
votes.
As a result the first-stage `close_wait` could return before every
source had sent CLOSE_LOAD. The early source then sent CLOSE_LOAD on its
incremental streams while the late sources had not opened theirs yet. A
destination's incremental `LoadStream` counts senders dynamically
(`add_source()` bumps `_total_streams`), so it concluded that all
senders had closed and committed the tablet early, which surfaces as
double close and segment number mismatch errors.
This PR excludes streams that have not entered the current close stage
from V2 quorum accounting (`!stream->is_closing()`), matching the V1
participant semantics while preserving the two-stage close algorithm. V1
already has the equivalent guard: `IndexChannel::_quorum_success()`
skips a node channel when `check_status()` fails, and `check_status()`
is `none_of({_cancelled, !_eos_is_produced})`, i.e. it also requires the
channel to have been `mark_close()`d for the current stage.
Also documents why leaving the first-stage `close_wait` on quorum
success still preserves the cross-source fence.
Tests:
-
`TestVTabletWriterV2.quorum_excludes_streams_not_closing_in_current_stage`
- `auto_partition_quorum_race_docker` (20 attempts, 1 FE + 3 BE,
`enable_quorum_success_write=true`)
---
be/src/exec/sink/load_stream_stub.h | 2 +
be/src/exec/sink/writer/vtablet_writer_v2.cpp | 24 +++++-
be/test/exec/sink/vtablet_writer_v2_test.cpp | 28 +++++++
.../sql/auto_partition_quorum_race_docker.groovy | 98 ++++++++++++++++++++++
4 files changed, 148 insertions(+), 4 deletions(-)
diff --git a/be/src/exec/sink/load_stream_stub.h
b/be/src/exec/sink/load_stream_stub.h
index a6699ba639e..8951c483220 100644
--- a/be/src/exec/sink/load_stream_stub.h
+++ b/be/src/exec/sink/load_stream_stub.h
@@ -215,6 +215,8 @@ public:
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/exec/sink/writer/vtablet_writer_v2.cpp
b/be/src/exec/sink/writer/vtablet_writer_v2.cpp
index 2db1002d020..7eb0138beb7 100644
--- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp
+++ b/be/src/exec/sink/writer/vtablet_writer_v2.cpp
@@ -726,9 +726,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.
@@ -898,7 +912,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;
}
diff --git a/be/test/exec/sink/vtablet_writer_v2_test.cpp
b/be/test/exec/sink/vtablet_writer_v2_test.cpp
index 831fa613e72..759ca06b51f 100644
--- a/be/test/exec/sink/vtablet_writer_v2_test.cpp
+++ b/be/test/exec/sink/vtablet_writer_v2_test.cpp
@@ -492,4 +492,32 @@ TEST_F(TestVTabletWriterV2, fail_two_miss_one_same_tablet)
{
ASSERT_EQ(st, Status::InternalError("test"));
}
+TEST_F(TestVTabletWriterV2,
quorum_excludes_streams_not_closing_in_current_stage) {
+ UniqueId load_id;
+ auto load_stream_map = std::make_shared<LoadStreamMap>(load_id, src_id, 1,
1, nullptr);
+ auto initial_streams = load_stream_map->get_or_create(1001);
+ auto incremental_streams_1 = load_stream_map->get_or_create(1002, true);
+ auto incremental_streams_2 = load_stream_map->get_or_create(1003, true);
+
+ auto writer = create_vtablet_writer();
+ writer->_load_stream_map = load_stream_map;
+ writer->_tablets_by_node[1002].insert(1);
+ writer->_tablets_by_node[1003].insert(1);
+
+ std::unordered_set<std::shared_ptr<LoadStreamStub>> unfinished_streams {
+ initial_streams->streams().front()};
+ std::unordered_set<int64_t> need_finish_tablets {1};
+
+ // The incremental streams do not participate in the first close stage.
They must not be
+ // treated as finished just because they are absent from
unfinished_streams.
+ ASSERT_FALSE(writer->_quorum_success(unfinished_streams,
need_finish_tablets));
+
+ // Once the incremental streams participate in the second stage and
finish, they can satisfy
+ // quorum normally.
+ incremental_streams_1->streams().front()->_is_closing.store(true);
+ incremental_streams_2->streams().front()->_is_closing.store(true);
+ unfinished_streams.clear();
+ ASSERT_TRUE(writer->_quorum_success(unfinished_streams,
need_finish_tablets));
+}
+
} // namespace doris
diff --git
a/regression-test/suites/partition_p1/auto_partition/sql/auto_partition_quorum_race_docker.groovy
b/regression-test/suites/partition_p1/auto_partition/sql/auto_partition_quorum_race_docker.groovy
new file mode 100644
index 00000000000..f50670069af
--- /dev/null
+++
b/regression-test/suites/partition_p1/auto_partition/sql/auto_partition_quorum_race_docker.groovy
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("auto_partition_quorum_race_docker", "docker") {
+ def options = new ClusterOptions()
+ options.cloudMode = false
+ options.feNum = 1
+ options.beNum = 3
+ options.beConfigs += [
+ "enable_quorum_success_write=true"
+ ]
+
+ docker(options) {
+ // Exercise the normal scheduler. No preferred key, affinity mode, or
+ // non-default backend location tag participates in this test.
+ sql "set enable_load_backend_selection = false"
+ sql "set preferred_backend_selection_key = ''"
+ sql "set backend_selection_mode = 'default'"
+
+ assertEquals("false",
+ sql("show variables like
'enable_load_backend_selection'")[0][1].toString())
+ assertEquals("",
+ sql("show variables like
'preferred_backend_selection_key'")[0][1].toString())
+ assertEquals("default",
+ sql("show variables like
'backend_selection_mode'")[0][1].toString())
+
+ sql "drop table if exists auto_partition_race_src force"
+ sql """
+ create table auto_partition_race_src (
+ k0 date not null
+ )
+ distributed by hash(k0) buckets 2
+ properties("replication_num" = "1")
+ """
+
+ // The two buckets are scanned by two sink instances. Their completion
+ // times intentionally differ, matching the original DORIS-27500 case.
+ sql """insert into auto_partition_race_src values ("2012-12-11")"""
+ sql """
+ insert into auto_partition_race_src
+ select "2020-12-12" from numbers("number" = "20000")
+ """
+
+ def sourceTablets = sql_return_maparray("show tablets from
auto_partition_race_src")
+ def sourceBackendIds = sourceTablets.collect { it.BackendId.toString()
}.toSet()
+ assertEquals(2, sourceTablets.size())
+ assertEquals(2, sourceBackendIds.size(),
+ "source tablets must reside on two BEs to create two remote
sink instances")
+
+ int attempts = 20
+ int failures = 0
+ def failureMessages = []
+ for (int attempt = 1; attempt <= attempts; ++attempt) {
+ def tableName = "auto_partition_race_dst_${attempt}"
+ sql "drop table if exists ${tableName} force"
+ sql """
+ create table ${tableName} (
+ k0 date not null
+ )
+ auto partition by range (date_trunc(k0, 'day')) ()
+ distributed by hash(k0) buckets 10
+ properties("replication_num" = "3")
+ """
+
+ try {
+ sql "insert into ${tableName} select * from
auto_partition_race_src"
+ assertEquals(20001L, sql("select count(*) from
${tableName}")[0][0] as long)
+ } catch (Exception e) {
+ ++failures
+ failureMessages.add("attempt ${attempt}: ${e.getMessage()}")
+ logger.warn("DORIS-27500 reproduced on attempt ${attempt}:
${e.getMessage()}")
+ } finally {
+ sql "drop table if exists ${tableName} force"
+ }
+ }
+
+ logger.info("DORIS-27500 reproduction result: ${failures}/${attempts}
failed")
+ assertEquals(0, failures,
+ "ordinary auto partition loads failed without affinity:\n"
+ + failureMessages.join("\n"))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
