Backport #89652 to 25.10: Add runtime filters only for supported join algorithms

robot-clickhouse · robot-clickhouse · commit d3b10a3e365e · 2025-11-06T21:12:08.000Z
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.reference b/tests/queries/0_stateless/03580_join_runtime_filter.reference
@@ -8,6 +8,7 @@
 500
 6000
 6000
+Algorithm: FullSortingMergeJoin
 Filter column: and(__filterContains(_runtime_filter_UNIQ_ID_0, __table1.c_nationkey), __filterContains(_runtime_filter_UNIQ_ID_1, modulo(__table1.c_custkey, 100_UInt8))) (removed)
 BuildRuntimeFilter (Build runtime join filter on modulo(__table2.n_nationkey, 10_UInt8) (_runtime_filter_UNIQ_ID_1))
 BuildRuntimeFilter (Build runtime join filter on __table2.n_nationkey (_runtime_filter_UNIQ_ID_0))
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
@@ -73,6 +74,17 @@ FROM customer, nation
 WHERE c_nationkey = n_nationkey AND n_name = 'FRANCE'
 SETTINGS enable_join_runtime_filters = 1, join_runtime_bloom_filter_bytes = 4096, join_runtime_bloom_filter_hash_functions = 2;
 
+-- Join algorithm that doesn't support runtime filters
+SELECT REGEXP_REPLACE(trimLeft(explain), '_runtime_filter_\\d+', '_runtime_filter_UNIQ_ID')
+FROM (
+    EXPLAIN actions=1
+    SELECT count()
+    FROM customer, nation
+    WHERE c_nationkey = n_nationkey AND n_nationkey%10 = c_custkey%100
+    SETTINGS query_plan_join_swap_table=0, join_algorithm='full_sorting_merge'
+)
+WHERE (explain ILIKE '%Filter column%') OR (explain LIKE '%BuildRuntimeFilter%') OR (explain LIKE '%FullSorting%');
+
 -- Filters on multiple join predicates
 SELECT REGEXP_REPLACE(trimLeft(explain), '_runtime_filter_\\d+', '_runtime_filter_UNIQ_ID')
 FROM (
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')

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@`
`8`	`8`	`500`
`9`	`9`	`6000`
`10`	`10`	`6000`
	`11`	`+Algorithm: FullSortingMergeJoin`
`11`	`12`	`Filter column: and(__filterContains(_runtime_filter_UNIQ_ID_0, __table1.c_nationkey), __filterContains(_runtime_filter_UNIQ_ID_1, modulo(__table1.c_custkey, 100_UInt8))) (removed)`
`12`	`13`	`BuildRuntimeFilter (Build runtime join filter on modulo(__table2.n_nationkey, 10_UInt8) (_runtime_filter_UNIQ_ID_1))`
`13`	`14`	`BuildRuntimeFilter (Build runtime join filter on __table2.n_nationkey (_runtime_filter_UNIQ_ID_0))`