This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 4d3d5024789 branch-4.1: [fix](load) correct quorum participants for
incremental streams #66016 (#66171)
4d3d5024789 is described below
commit 4d3d5024789575438d360ffc29cef3fcf83045f9
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jul 30 08:49:34 2026 +0800
branch-4.1: [fix](load) correct quorum participants for incremental streams
#66016 (#66171)
Cherry-picked from #66016
Co-authored-by: deardeng <[email protected]>
---
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 638d82882a7..21b0dc9b404 100644
--- a/be/src/exec/sink/load_stream_stub.h
+++ b/be/src/exec/sink/load_stream_stub.h
@@ -216,6 +216,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 ce78cb3b270..cb6ec89ff48 100644
--- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp
+++ b/be/src/exec/sink/writer/vtablet_writer_v2.cpp
@@ -727,9 +727,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.
@@ -899,7 +913,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 10814c4b9db..6a508a9ef23 100644
--- a/be/test/exec/sink/vtablet_writer_v2_test.cpp
+++ b/be/test/exec/sink/vtablet_writer_v2_test.cpp
@@ -505,4 +505,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]
