You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by pa...@apache.org on 2026/04/15 14:09:31 UTC
(doris) branch master updated: [fix](runtime-filter) Fix race condition in Set operator runtime filter processing (#62434)
This is an automated email from the ASF dual-hosted git repository.
panxiaolei 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 c9e6e798055 [fix](runtime-filter) Fix race condition in Set operator runtime filter processing (#62434)
c9e6e798055 is described below
commit c9e6e798055696bcb5c1fe17f9bdd5ac80123993
Author: Pxl <xl...@selectdb.com>
AuthorDate: Wed Apr 15 22:09:22 2026 +0800
[fix](runtime-filter) Fix race condition in Set operator runtime filter processing (#62434)
This pull request addresses a race condition in the Set operator
(`INTERSECT`/`EXCEPT`) when using the `IN_FILTER` runtime filter type.
The main fix ensures that the hash table size used for runtime filter
processing is captured before the probe side can modify the hash table,
preventing errors during query execution. Additionally, regression tests
are added to verify the fix and prevent future regressions.
**Fixes for race condition in Set operator:**
* The hash table size is now snapshotted in
`SetSinkOperatorX<is_intersect>::sink()` before enabling the probe
pipeline, and this saved value is used for runtime filter processing in
`SetSinkLocalState<is_intersect>::close()`. This prevents the probe side
from shrinking the hash table before the runtime filter is processed,
avoiding possible `InternalError` failures.
[[1]](diffhunk://#diff-7658815d4a9a8268b4ef4ad00a93998af07a421b538ad608e3f3a0e20ef805cdR103-R107)
[[2]](diffhunk://#diff-7658815d4a9a8268b4ef4ad00a93998af07a421b538ad608e3f3a0e20ef805cdL48-R47)
[[3]](diffhunk://#diff-1a8671c9fbc97801353bffeffe99e2189a16d275d3f865714e2cc6750ee8c54fR65-R68)
* Added a new member variable `_build_hash_table_size` to
`SetSinkLocalState` to store the snapshotted hash table size.
**Testing and regression coverage:**
* Added a new regression test suite `set_operator_in_filter.groovy` and
its output file to reproduce and verify the race condition, ensuring the
fix is tested for both `INTERSECT` and `EXCEPT` queries with `IN_FILTER`
type and small `max_in_num`.
[[1]](diffhunk://#diff-2ff8fca2d88b291375196153f6f9377e6a0d80a2b8222d63ef67d2dc307cb332R1-R144)
[[2]](diffhunk://#diff-795c4c2d53c199ee4187b5aef698e62a6c993d2aad99014063c0c3540553caf3R1-R23)
**Code cleanup:**
* Removed unnecessary `#include "common/compile_check_begin.h"` and
`#include "common/compile_check_end.h"` lines from several files for
better code clarity.
[[1]](diffhunk://#diff-7658815d4a9a8268b4ef4ad00a93998af07a421b538ad608e3f3a0e20ef805cdL27)
[[2]](diffhunk://#diff-90e895f6474975cbb1d9b2fc276bc6d9e1a22675f5126b836fb83c13fd148eb5L31)
[[3]](diffhunk://#diff-90e895f6474975cbb1d9b2fc276bc6d9e1a22675f5126b836fb83c13fd148eb5L57)
[[4]](diffhunk://#diff-a8e4192eb0bf59708c56d0a59c46f601684a554628105f70c50211784a1c5271L26)
---------
Co-authored-by: Copilot <22...@users.noreply.github.com>
---
be/src/exec/operator/set_sink_operator.cpp | 7 +-
be/src/exec/operator/set_sink_operator.h | 5 +
.../runtime_filter/set_operator_in_filter.out | 23 ++++
.../runtime_filter/set_operator_in_filter.groovy | 138 +++++++++++++++++++++
4 files changed, 172 insertions(+), 1 deletion(-)
diff --git a/be/src/exec/operator/set_sink_operator.cpp b/be/src/exec/operator/set_sink_operator.cpp
index ba81d0e1b67..ec037697e98 100644
--- a/be/src/exec/operator/set_sink_operator.cpp
+++ b/be/src/exec/operator/set_sink_operator.cpp
@@ -44,7 +44,7 @@ Status SetSinkLocalState<is_intersect>::close(RuntimeState* state, Status exec_s
if (!_terminated && _runtime_filter_producer_helper && !state->is_cancelled()) {
try {
RETURN_IF_ERROR(_runtime_filter_producer_helper->process(
- state, &_shared_state->build_block, _shared_state->get_hash_table_size()));
+ state, &_shared_state->build_block, _build_hash_table_size));
} catch (Exception& e) {
return Status::InternalError(
"rf process meet error: {}, _terminated: {}, _finish_dependency: {}",
@@ -100,6 +100,11 @@ Status SetSinkOperatorX<is_intersect>::sink(RuntimeState* state, Block* in_block
COUNTER_SET(local_state._hash_table_size, (int64_t)hash_table_size);
COUNTER_SET(local_state._valid_element_in_hash_table, valid_element_in_hash_tbl);
+ // Snapshot hash table size before enabling the probe pipeline. The probe side
+ // can shrink the hash table via _refresh_hash_table() after set_ready(), so
+ // close() must use this saved value for runtime filter processing.
+ local_state._build_hash_table_size = hash_table_size;
+
local_state._shared_state->probe_finished_children_dependency[_cur_child_id + 1]
->set_ready();
DCHECK_GT(_child_quantity, 1);
diff --git a/be/src/exec/operator/set_sink_operator.h b/be/src/exec/operator/set_sink_operator.h
index ff76ea2f0a3..e1c937f1471 100644
--- a/be/src/exec/operator/set_sink_operator.h
+++ b/be/src/exec/operator/set_sink_operator.h
@@ -45,6 +45,7 @@ public:
Status open(RuntimeState* state) override;
Status terminate(RuntimeState* state) override;
Status close(RuntimeState* state, Status exec_status) override;
+ Dependency* finishdependency() override { return _finish_dependency.get(); }
private:
friend class SetSinkOperatorX<is_intersect>;
@@ -60,6 +61,10 @@ private:
std::shared_ptr<RuntimeFilterProducerHelperSet> _runtime_filter_producer_helper;
std::shared_ptr<CountedFinishDependency> _finish_dependency;
+ // Snapshot of hash table size taken in sink(eos) before set_ready(). The probe side can
+ // modify the hash table via _refresh_hash_table() after set_ready(), so close() must use
+ // this saved value instead of calling get_hash_table_size() again.
+ uint64_t _build_hash_table_size = 0;
};
template <bool is_intersect>
diff --git a/regression-test/data/query_p0/runtime_filter/set_operator_in_filter.out b/regression-test/data/query_p0/runtime_filter/set_operator_in_filter.out
new file mode 100644
index 00000000000..b0d2a4133b5
--- /dev/null
+++ b/regression-test/data/query_p0/runtime_filter/set_operator_in_filter.out
@@ -0,0 +1,23 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !intersect_in_rf --
+1
+11
+13
+15
+3
+5
+7
+9
+
+-- !except_in_rf --
+
+-- !intersect_in_rf_zero --
+1
+11
+13
+15
+3
+5
+7
+9
+
diff --git a/regression-test/suites/query_p0/runtime_filter/set_operator_in_filter.groovy b/regression-test/suites/query_p0/runtime_filter/set_operator_in_filter.groovy
new file mode 100644
index 00000000000..c44f3caba1c
--- /dev/null
+++ b/regression-test/suites/query_p0/runtime_filter/set_operator_in_filter.groovy
@@ -0,0 +1,138 @@
+// 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.
+
+// Regression test for race condition in Set operator (INTERSECT/EXCEPT) with
+// IN_FILTER runtime filter type.
+//
+// Before the fix, when runtime_filter_type=1 (IN) and runtime_filter_max_in_num
+// was small, a race between the build sink's close() and the probe's hash table
+// refresh could cause RuntimeFilterWrapper::insert() to fail with InternalError.
+// The fix snapshots hash_table_size in sink(eos) before set_ready() and uses the
+// saved value in close() for RF processing. Also overrides finishdependency() so
+// the pipeline task properly waits for sync_filter_size completion.
+
+suite("set_operator_in_filter") {
+ sql "DROP TABLE IF EXISTS set_rf_t1"
+ sql "DROP TABLE IF EXISTS set_rf_t2"
+ sql "DROP TABLE IF EXISTS set_rf_t3"
+
+ sql """
+ CREATE TABLE set_rf_t1 (
+ k1 INT NOT NULL,
+ v1 VARCHAR(64) NOT NULL
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ PROPERTIES("replication_num" = "1");
+ """
+
+ sql """
+ CREATE TABLE set_rf_t2 (
+ k1 INT NOT NULL,
+ v1 VARCHAR(64) NOT NULL
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ PROPERTIES("replication_num" = "1");
+ """
+
+ sql """
+ CREATE TABLE set_rf_t3 (
+ k1 INT NOT NULL,
+ v1 VARCHAR(64) NOT NULL
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ PROPERTIES("replication_num" = "1");
+ """
+
+ // Insert >10 unique values into t1 so that with max_in_num=10 the IN filter
+ // would overflow. This is the key condition that triggers the original bug.
+ sql """
+ INSERT INTO set_rf_t1 VALUES
+ (1,'a'),(2,'b'),(3,'c'),(4,'d'),(5,'e'),
+ (6,'f'),(7,'g'),(8,'h'),(9,'i'),(10,'j'),
+ (11,'k'),(12,'l'),(13,'m'),(14,'n'),(15,'o'),
+ (16,'p'),(17,'q'),(18,'r'),(19,'s'),(20,'t');
+ """
+
+ // t2 shares some values with t1 (for INTERSECT) and has extras (for EXCEPT)
+ sql """
+ INSERT INTO set_rf_t2 VALUES
+ (1,'a'),(3,'c'),(5,'e'),(7,'g'),(9,'i'),
+ (11,'k'),(13,'m'),(15,'o'),(17,'q'),(19,'s'),
+ (21,'u'),(22,'v');
+ """
+
+ // t3 is the probe-side table for the join that triggers RF generation
+ sql """
+ INSERT INTO set_rf_t3 VALUES
+ (1,'x'),(3,'x'),(5,'x'),(7,'x'),(9,'x'),
+ (11,'x'),(13,'x'),(15,'x'),(100,'x'),(200,'x');
+ """
+
+ sql "sync"
+
+ // Use pure IN_FILTER type with a small max_in_num to reproduce the conditions
+ // that triggered the race condition bug.
+ sql "set runtime_filter_type = 1"
+ sql "set runtime_filter_max_in_num = 10"
+ sql "set disable_join_reorder = true"
+
+ // INTERSECT with join: the INTERSECT probe shrinks the hash table, which
+ // previously could race with RF processing in the build sink's close().
+ order_qt_intersect_in_rf """
+ SELECT T.k1 FROM (
+ SELECT k1 FROM set_rf_t1
+ INTERSECT
+ SELECT k1 FROM set_rf_t2
+ ) T JOIN set_rf_t3 ON T.k1 = set_rf_t3.k1;
+ """
+
+ // EXCEPT with join: similar scenario but with EXCEPT semantics.
+ order_qt_except_in_rf """
+ SELECT T.k1 FROM (
+ SELECT k1 FROM set_rf_t1
+ EXCEPT
+ SELECT k1 FROM set_rf_t2
+ ) T JOIN set_rf_t3 ON T.k1 = set_rf_t3.k1;
+ """
+
+ // Run INTERSECT query multiple times to increase coverage of the timing
+ // window. Before the fix, this would intermittently fail.
+ for (int i = 0; i < 5; i++) {
+ sql """
+ SELECT T.k1 FROM (
+ SELECT k1 FROM set_rf_t1
+ INTERSECT
+ SELECT k1 FROM set_rf_t2
+ ) T JOIN set_rf_t3 ON T.k1 = set_rf_t3.k1
+ ORDER BY T.k1;
+ """
+ }
+
+ // Also test with max_in_num=0 (always overflow immediately)
+ sql "set runtime_filter_max_in_num = 0"
+
+ order_qt_intersect_in_rf_zero """
+ SELECT T.k1 FROM (
+ SELECT k1 FROM set_rf_t1
+ INTERSECT
+ SELECT k1 FROM set_rf_t2
+ ) T JOIN set_rf_t3 ON T.k1 = set_rf_t3.k1;
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org