branch-4.2 [feature](runtime filter) Add decoupled runtime filter support

be/src/exec/operator/hashjoin_build_sink.cpp

@@ -103,8 +103,8 @@ Status HashJoinBuildSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo
 
     _runtime_filter_producer_helper = std::make_shared<RuntimeFilterProducerHelper>(
             _should_build_hash_table, p._is_broadcast_join);
-    RETURN_IF_ERROR(_runtime_filter_producer_helper->init(state, _build_expr_ctxs,
-                                                          p._runtime_filter_descs));
+    RETURN_IF_ERROR(_runtime_filter_producer_helper->init(
+            state, _build_expr_ctxs, p._runtime_filter_descs, p._child->row_desc()));
     return Status::OK();
 }
 

be/src/exec/runtime_filter/runtime_filter.cpp

@@ -79,6 +79,9 @@ Status RuntimeFilter::_init_with_desc(const TRuntimeFilterDesc* desc,
                                       const TQueryOptions* options) {
     VExprContextSPtr build_ctx;
     RETURN_IF_ERROR(VExpr::create_expr_tree(desc->src_expr, build_ctx));
+    if (build_ctx == nullptr) {
+        return Status::InternalError("runtime filter {} has empty src_expr", desc->filter_id);
+    }
 
     RuntimeFilterParams params;
     params.filter_id = desc->filter_id;

be/src/exec/runtime_filter/runtime_filter_consumer.h

@@ -91,10 +91,16 @@ class RuntimeFilterConsumer : public RuntimeFilter {
               _probe_expr(desc->planId_to_target_expr.find(node_id)->second),
               _registration_time(MonotonicMillis()),
               _rf_state(State::NOT_READY) {
-        auto* query_ctx = state->get_query_ctx();
-        bool wait_infinitely = state->runtime_filter_wait_infinitely() || !has_remote_target();
-        _rf_wait_time_ms = wait_infinitely ? query_ctx->execution_timeout() * 1000
-                                           : query_ctx->runtime_filter_wait_time_ms();
+        if (desc->__isset.wait_time_ms) {
+            _rf_wait_time_ms = desc->wait_time_ms;
+        } else {
+            // local rf must wait until timeout, otherwise it may lead results incorrectness,
+            // because LEFT_SEMI_DIRECT_RETURN_OPT
+            auto* query_ctx = state->get_query_ctx();
+            bool wait_infinitely = state->runtime_filter_wait_infinitely() || !has_remote_target();
+            _rf_wait_time_ms = wait_infinitely ? query_ctx->execution_timeout() * 1000
+                                               : query_ctx->runtime_filter_wait_time_ms();
+        }
         DorisMetrics::instance()->runtime_filter_consumer_num->increment(1);
     }
 

be/src/exec/runtime_filter/runtime_filter_producer_helper.cpp

@@ -21,27 +21,67 @@
 
 #include "exec/pipeline/pipeline_task.h"
 #include "exec/runtime_filter/runtime_filter_wrapper.h"
+#include "exprs/vexpr.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
-void RuntimeFilterProducerHelper::_init_expr(
+Status RuntimeFilterProducerHelper::_init_expr(
         const VExprContextSPtrs& build_expr_ctxs,
         const std::vector<TRuntimeFilterDesc>& runtime_filter_descs) {
     _filter_expr_contexts.resize(runtime_filter_descs.size());
+    _decoupled_filter_indices.clear();
     for (size_t i = 0; i < runtime_filter_descs.size(); i++) {
-        _filter_expr_contexts[i] = build_expr_ctxs[runtime_filter_descs[i].expr_order];
+        if (runtime_filter_descs[i].expr_order >= 0) {
+            _filter_expr_contexts[i] = build_expr_ctxs[runtime_filter_descs[i].expr_order];
+        } else {
+            // Decoupled RF: srcExpr is not in the builder's equi-conjuncts.
+            // Create a new VExprContext from the src_expr in the thrift descriptor.
+            VExprContextSPtr ctx;
+            RETURN_IF_ERROR(VExpr::create_expr_tree(runtime_filter_descs[i].src_expr, ctx));
+            if (ctx == nullptr) {
+                return Status::InternalError("decoupled runtime filter {} has empty src_expr",
+                                             runtime_filter_descs[i].filter_id);
+            }
+            _filter_expr_contexts[i] = std::move(ctx);
+            _decoupled_filter_indices.push_back(i);
+        }
     }
+    return Status::OK();
 }
 
 Status RuntimeFilterProducerHelper::init(
         RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
         const std::vector<TRuntimeFilterDesc>& runtime_filter_descs) {
+    return _init(state, build_expr_ctxs, runtime_filter_descs, nullptr);
+}
+
+Status RuntimeFilterProducerHelper::init(
+        RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
+        const std::vector<TRuntimeFilterDesc>& runtime_filter_descs,
+        const RowDescriptor& row_desc) {
+    return _init(state, build_expr_ctxs, runtime_filter_descs, &row_desc);
+}
+
+Status RuntimeFilterProducerHelper::_init(
+        RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
+        const std::vector<TRuntimeFilterDesc>& runtime_filter_descs,
+        const RowDescriptor* row_desc) {
     _producers.resize(runtime_filter_descs.size());
     for (size_t i = 0; i < runtime_filter_descs.size(); i++) {
         RETURN_IF_ERROR(
                 state->register_producer_runtime_filter(runtime_filter_descs[i], &_producers[i]));
     }
-    _init_expr(build_expr_ctxs, runtime_filter_descs);
+    RETURN_IF_ERROR(_init_expr(build_expr_ctxs, runtime_filter_descs));
+    if (_decoupled_filter_indices.empty()) {
+        return Status::OK();
+    }
+    if (row_desc == nullptr) {
+        return Status::InternalError("decoupled runtime filters require row_desc during init");
+    }
+    for (size_t idx : _decoupled_filter_indices) {
+        RETURN_IF_ERROR(_filter_expr_contexts[idx]->prepare(state, *row_desc));
+        RETURN_IF_ERROR(_filter_expr_contexts[idx]->open(state));
+    }
     return Status::OK();
 }
 
@@ -90,7 +130,7 @@ Status RuntimeFilterProducerHelper::_publish(RuntimeState* state) {
 }
 
 Status RuntimeFilterProducerHelper::build(
-        RuntimeState* state, const Block* block, bool use_shared_table,
+        RuntimeState* state, Block* block, bool use_shared_table,
         std::map<int, std::shared_ptr<RuntimeFilterWrapper>>& runtime_filters) {
     if (_skip_runtime_filters_process) {
         return Status::OK();
@@ -101,6 +141,19 @@ Status RuntimeFilterProducerHelper::build(
         uint64_t hash_table_size = block ? block->rows() : 0;
         RETURN_IF_ERROR(_init_filters(state, hash_table_size));
         if (hash_table_size > 1) {
+            // Evaluate decoupled RF expressions on the block before insert.
+            // Standard RF exprs are already evaluated during hash table building.
+            for (size_t idx : _decoupled_filter_indices) {
+                int result_column_id = -1;
+                RETURN_IF_ERROR(_filter_expr_contexts[idx]->execute(block, &result_column_id));
+                // Materialize ColumnConst to a full column before insert.
+                // A decoupled source expression like k * 0 may produce a
+                // ColumnConst (single value), but _insert() expects a
+                // full-length column when inserting starting at a non-zero offset.
+                block->get_by_position(result_column_id).column =
+                        block->get_by_position(result_column_id)
+                                .column->convert_to_full_column_if_const();
+            }
             constexpr int HASH_JOIN_INSERT_OFFSET = 1; // the first row is mocked on hash join sink
             RETURN_IF_ERROR(_insert(block, HASH_JOIN_INSERT_OFFSET));
         }

be/src/exec/runtime_filter/runtime_filter_producer_helper.h

@@ -26,6 +26,7 @@
 #include "exec/runtime_filter/runtime_filter_mgr.h"
 #include "exec/runtime_filter/runtime_filter_producer.h"
 #include "exprs/vexpr_context.h"
+#include "runtime/descriptors.h"
 #include "runtime/runtime_state.h"
 
 namespace doris {
@@ -49,6 +50,9 @@ class RuntimeFilterProducerHelper {
     // create and register runtime filters producers
     Status init(RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
                 const std::vector<TRuntimeFilterDesc>& runtime_filter_descs);
+    Status init(RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
+                const std::vector<TRuntimeFilterDesc>& runtime_filter_descs,
+                const RowDescriptor& row_desc);
 
     // send local size to remote to sync global rf size if needed
     MOCK_FUNCTION Status
@@ -59,7 +63,7 @@ class RuntimeFilterProducerHelper {
     MOCK_FUNCTION Status skip_process(RuntimeState* state);
 
     // build rf
-    Status build(RuntimeState* state, const Block* block, bool use_shared_table,
+    Status build(RuntimeState* state, Block* block, bool use_shared_table,
                  std::map<int, std::shared_ptr<RuntimeFilterWrapper>>& runtime_filters);
 
     // publish rf
@@ -70,8 +74,11 @@ class RuntimeFilterProducerHelper {
     std::shared_ptr<RuntimeFilterWrapper> detect_local_in_filter(RuntimeState* state);
 
 protected:
-    virtual void _init_expr(const VExprContextSPtrs& build_expr_ctxs,
-                            const std::vector<TRuntimeFilterDesc>& runtime_filter_descs);
+    virtual Status _init_expr(const VExprContextSPtrs& build_expr_ctxs,
+                              const std::vector<TRuntimeFilterDesc>& runtime_filter_descs);
+    Status _init(RuntimeState* state, const VExprContextSPtrs& build_expr_ctxs,
+                 const std::vector<TRuntimeFilterDesc>& runtime_filter_descs,
+                 const RowDescriptor* row_desc);
     Status _init_filters(RuntimeState* state, uint64_t local_hash_table_size);
     Status _insert(const Block* block, size_t start);
     Status _publish(RuntimeState* state);
@@ -89,6 +96,9 @@ class RuntimeFilterProducerHelper {
     const bool _is_broadcast_join;
 
     std::vector<std::shared_ptr<VExprContext>> _filter_expr_contexts;
+    // Indices of _filter_expr_contexts that correspond to decoupled RFs (expr_order == -1).
+    // These need explicit evaluate on the build block before _insert().
+    std::vector<size_t> _decoupled_filter_indices;
 };
 #include "common/compile_check_end.h"
 } // namespace doris

be/src/exec/runtime_filter/runtime_filter_producer_helper_cross.h

@@ -60,9 +60,10 @@ class RuntimeFilterProducerHelperCross : public RuntimeFilterProducerHelper {
         return Status::OK();
     }
 
-    void _init_expr(const VExprContextSPtrs& build_expr_ctxs,
-                    const std::vector<TRuntimeFilterDesc>& runtime_filter_descs) override {
+    Status _init_expr(const VExprContextSPtrs& build_expr_ctxs,
+                      const std::vector<TRuntimeFilterDesc>& runtime_filter_descs) override {
         _filter_expr_contexts = build_expr_ctxs;
+        return Status::OK();
     }
 };
 #include "common/compile_check_end.h"

be/test/exec/runtime_filter/runtime_filter_producer_helper_test.cpp

@@ -55,6 +55,15 @@ class RuntimeFilterProducerHelperTest : public RuntimeFilterTest {
     ObjectPool _pool;
 };
 
+class RuntimeFilterProducerHelperForTest : public RuntimeFilterProducerHelper {
+public:
+    RuntimeFilterProducerHelperForTest() : RuntimeFilterProducerHelper(true, false) {}
+
+    Status init_expr_for_test(const std::vector<TRuntimeFilterDesc>& runtime_filter_descs) {
+        return _init_expr({}, runtime_filter_descs);
+    }
+};
+
 TEST_F(RuntimeFilterProducerHelperTest, basic) {
     auto helper = RuntimeFilterProducerHelper(true, false);
 
@@ -80,6 +89,71 @@ TEST_F(RuntimeFilterProducerHelperTest, basic) {
     FAIL_IF_ERROR_OR_CATCH_EXCEPTION(helper.publish(_runtime_states[0].get()));
 }
 
+TEST_F(RuntimeFilterProducerHelperTest, decoupled_filter_init) {
+    auto helper = RuntimeFilterProducerHelper(true, false);
+
+    std::vector<TRuntimeFilterDesc> runtime_filter_descs = {
+            TRuntimeFilterDescBuilder(0, TRuntimeFilterDescBuilder::get_default_expr(), -1,
+                                      {{0, TRuntimeFilterDescBuilder::get_default_expr()}})
+                    .build()};
+
+    SlotDescriptor slot_desc;
+    slot_desc._type = DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_INT, false);
+    TupleDescriptor tuple_desc;
+    tuple_desc.add_slot(&slot_desc);
+    RowDescriptor row_desc;
+    _tbl._slot_desc_map[0] = &slot_desc;
+    const_cast<std::vector<TupleDescriptor*>&>(row_desc._tuple_desc_map).push_back(&tuple_desc);
+
+    FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+            helper.init(_runtime_states[0].get(), {}, runtime_filter_descs, row_desc));
+
+    Block block;
+    auto column = ColumnInt32::create();
+    column->insert(Field::create_field<TYPE_INT>(1));
+    column->insert(Field::create_field<TYPE_INT>(2));
+    block.insert({std::move(column), std::make_shared<DataTypeInt32>(), "col1"});
+
+    std::map<int, std::shared_ptr<RuntimeFilterWrapper>> runtime_filters;
+    FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+            helper.build(_runtime_states[0].get(), &block, false, runtime_filters));
+    FAIL_IF_ERROR_OR_CATCH_EXCEPTION(helper.publish(_runtime_states[0].get()));
+}
+
+TEST_F(RuntimeFilterProducerHelperTest, decoupled_filter_init_with_empty_expr) {
+    auto helper = RuntimeFilterProducerHelper(true, false);
+
+    std::vector<TRuntimeFilterDesc> runtime_filter_descs = {
+            TRuntimeFilterDescBuilder(0, TExpr(), -1,
+                                      {{0, TRuntimeFilterDescBuilder::get_default_expr()}})
+                    .build()};
+
+    SlotDescriptor slot_desc;
+    slot_desc._type = DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_INT, false);
+    TupleDescriptor tuple_desc;
+    tuple_desc.add_slot(&slot_desc);
+    RowDescriptor row_desc;
+    _tbl._slot_desc_map[0] = &slot_desc;
+    const_cast<std::vector<TupleDescriptor*>&>(row_desc._tuple_desc_map).push_back(&tuple_desc);
+
+    auto st = helper.init(_runtime_states[0].get(), {}, runtime_filter_descs, row_desc);
+    ASSERT_FALSE(st.ok());
+    ASSERT_TRUE(st.to_string().find("empty src_expr") != std::string::npos);
+}
+
+TEST_F(RuntimeFilterProducerHelperTest, decoupled_expr_init_with_empty_expr) {
+    RuntimeFilterProducerHelperForTest helper;
+
+    std::vector<TRuntimeFilterDesc> runtime_filter_descs = {
+            TRuntimeFilterDescBuilder(0, TExpr(), -1,
+                                      {{0, TRuntimeFilterDescBuilder::get_default_expr()}})
+                    .build()};
+
+    auto st = helper.init_expr_for_test(runtime_filter_descs);
+    ASSERT_FALSE(st.ok());
+    ASSERT_TRUE(st.to_string().find("empty src_expr") != std::string::npos);
+}
+
 TEST_F(RuntimeFilterProducerHelperTest, wake_up_eraly) {
     auto helper = RuntimeFilterProducerHelper(true, false);
 

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/RuntimeFilterTranslator.java

@@ -37,6 +37,7 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.statistics.StatisticalType;
+import org.apache.doris.thrift.TRuntimeFilterType;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -161,6 +162,7 @@ public void createLegacyRuntimeFilter(RuntimeFilter filter, JoinNodeBase node, P
                             scanNode, targetExpr, true, isLocalTarget));
                 }
                 origFilter.setBloomFilterSizeCalculatedByNdv(filter.isBloomFilterSizeCalculatedByNdv());
+                setWaitTimeMs(origFilter, filter.isNonBlocking(), isLocalTarget);
                 org.apache.doris.planner.RuntimeFilter finalizedFilter = finalize(origFilter);
                 scanNodeList.stream().filter(e -> e.getStatisticalType() == StatisticalType.CTE_SCAN_NODE)
                         .forEach(f -> {
@@ -186,4 +188,26 @@ private org.apache.doris.planner.RuntimeFilter finalize(org.apache.doris.planner
         origFilter.extractTargetsPosition();
         return origFilter;
     }
+
+    private void setWaitTimeMs(org.apache.doris.planner.RuntimeFilter filter,
+            boolean isNonBlocking, boolean isLocalTarget) {
+        if (isNonBlocking) {
+            filter.setWaitTimeMs(0);
+        } else {
+            if (ConnectContext.get() != null) {
+                SessionVariable sessionVar = ConnectContext.get().getSessionVariable();
+                if (sessionVar.runtimeFilterWaitInfinitely
+                        || filter.getType() == TRuntimeFilterType.BITMAP
+                        || isLocalTarget) {
+                    // wait infinitely
+                    filter.setWaitTimeMs(sessionVar.getQueryTimeoutS() * 1000);
+                } else {
+                    filter.setWaitTimeMs(sessionVar.getRuntimeFilterWaitTimeMs());
+                }
+            } else {
+                filter.setWaitTimeMs(1000);
+            }
+        }
+    }
+
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterContext.java

@@ -250,6 +250,35 @@ public void removeFilter(RuntimeFilter rf, ExprId targetId) {
         }
     }
 
+    /**
+     * Fully remove a runtime filter: remove all its targets, clean up from builder node,
+     * targetExprIdToFilter, and joinToTargetExprId maps.
+     */
+    public void removeRuntimeFilter(RuntimeFilter rf) {
+        // Remove from each target scan
+        for (int i = 0; i < rf.getTargetScans().size(); i++) {
+            rf.getTargetScans().get(i).removeAppliedRuntimeFilter(rf);
+        }
+        // Remove from targetExprIdToFilter
+        for (Slot slot : rf.getTargetSlots()) {
+            List<RuntimeFilter> filters = targetExprIdToFilter.get(slot.getExprId());
+            if (filters != null) {
+                filters.remove(rf);
+            }
+        }
+        // Remove from builder node
+        rf.getBuilderNode().getRuntimeFilters().remove(rf);
+        // Remove from joinToTargetExprId: remove target ExprIds that only belonged to this RF
+        for (Slot slot : rf.getTargetSlots()) {
+            for (Map.Entry<Plan, List<ExprId>> entry : joinToTargetExprId.entrySet()) {
+                if (entry.getKey().equals(rf.getBuilderNode())) {
+                    entry.getValue().remove(slot.getExprId());
+                }
+            }
+        }
+        prunedRF.add(rf);
+    }
+
     public void setTargetsOnScanNode(PhysicalRelation relation, Slot slot) {
         this.targetOnOlapScanNodeMap.computeIfAbsent(relation, k -> Lists.newArrayList()).add(slot);
     }