Add runtime filters only for supported join algorithms (i.e. hash joins)

davenger · davenger · commit 0d762d0b04e3 · 2025-11-06T15:28:57.000+01:00
diff --git a/src/Processors/QueryPlan/JoinStepLogical.h b/src/Processors/QueryPlan/JoinStepLogical.h
@@ -84,6 +84,7 @@ class JoinStepLogical final : public IQueryPlanStep
 
     const SortingStep::Settings & getSortingSettings() const { return sorting_settings; }
     const JoinSettings & getJoinSettings() const { return join_settings; }
+    JoinSettings & getJoinSettings() { return join_settings; }
     const JoinOperator & getJoinOperator() const { return join_operator; }
     JoinOperator & getJoinOperator() { return join_operator; }
 
diff --git a/src/Processors/QueryPlan/Optimizations/joinRuntimeFilter.cpp b/src/Processors/QueryPlan/Optimizations/joinRuntimeFilter.cpp
@@ -44,6 +44,14 @@ const ActionsDAG::Node & createRuntimeFilterCondition(ActionsDAG & actions_dag,
     return condition;
 }
 
+static bool supportsRuntimeFilter(JoinAlgorithm join_algorithm)
+{
+    /// Runtime filter can only be applied to join algorithms that first read the right side and only after that read the left side.
+    return
+        join_algorithm == JoinAlgorithm::HASH ||
+        join_algorithm == JoinAlgorithm::PARALLEL_HASH;
+}
+
 bool tryAddJoinRuntimeFilter(QueryPlan::Node & node, QueryPlan::Nodes & nodes, const QueryPlanOptimizationSettings & optimization_settings)
 {
     /// Is this a join step?
@@ -57,12 +65,14 @@ bool tryAddJoinRuntimeFilter(QueryPlan::Node & node, QueryPlan::Nodes & nodes, c
 
     /// Check if join can do runtime filtering on left table
     const auto & join_operator = join_step->getJoinOperator();
+    auto & join_algorithms = join_step->getJoinSettings().join_algorithms;
     const bool can_use_runtime_filter =
         (
             (join_operator.kind == JoinKind::Inner && (join_operator.strictness == JoinStrictness::All || join_operator.strictness == JoinStrictness::Any)) ||
             ((join_operator.kind == JoinKind::Left || join_operator.kind == JoinKind::Right) && join_operator.strictness == JoinStrictness::Semi)
         ) &&
-        (join_operator.locality == JoinLocality::Unspecified || join_operator.locality == JoinLocality::Local);
+        (join_operator.locality == JoinLocality::Unspecified || join_operator.locality == JoinLocality::Local) &&
+        std::find_if(join_algorithms.begin(), join_algorithms.end(), supportsRuntimeFilter) != join_algorithms.end();
 
     if (!can_use_runtime_filter)
         return false;
@@ -183,6 +193,9 @@ bool tryAddJoinRuntimeFilter(QueryPlan::Node & node, QueryPlan::Nodes & nodes, c
 
     node.children = {apply_filter_node, build_filter_node};
 
+    /// Remove algorithms that are not compatible with runtime filters
+    std::erase_if(join_algorithms, [](auto join_algorithm) { return !supportsRuntimeFilter(join_algorithm); });
+
     return true;
 }
 
diff --git a/tests/queries/0_stateless/03580_join_runtime_filter.sql b/tests/queries/0_stateless/03580_join_runtime_filter.sql
@@ -15,6 +15,7 @@ INSERT INTO orders SELECT number, number/10000, number%1000 FROM numbers(1000000
 SET enable_analyzer=1;
 SET enable_join_runtime_filters=1;
 SET enable_parallel_replicas=0;
+SET join_algorithm = 'hash,parallel_hash';
 
 SELECT avg(o_totalprice)
 FROM orders, customer, nation
diff --git a/tests/queries/0_stateless/03580_join_runtime_filter_prewhere.sql b/tests/queries/0_stateless/03580_join_runtime_filter_prewhere.sql
@@ -7,6 +7,7 @@ INSERT INTO customer SELECT number, 5 FROM numbers(500);
 
 SET enable_analyzer=1;
 SET enable_parallel_replicas=0;
+SET join_algorithm = 'hash,parallel_hash';
 
 SELECT '-- Check that filter on c_nationkey is moved to PREWHERE';
 SELECT REGEXP_REPLACE(trimLeft(explain), '_runtime_filter_\\d+', '_runtime_filter_UNIQ_ID')
