[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 <223556219+Copilot@users.noreply.github.com>

Author: BiteTheDDDDt

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);

be/src/exec/operator/set_sink_operator.h

@@ -45,6 +45,7 @@ class SetSinkLocalState final : public PipelineXSinkLocalState<SetSharedState> {
     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 @@ class SetSinkLocalState final : public PipelineXSinkLocalState<SetSharedState> {
 
     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>

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
+

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;
+    """
+}