Fix join filter pushdown

src/Analyzer/Utils.cpp

@@ -35,6 +35,7 @@
 #include <Storages/IStorage.h>
 
 #include <Interpreters/Context.h>
+#include <Interpreters/misc.h>
 
 #include <Analyzer/ArrayJoinNode.h>
 #include <Analyzer/ColumnNode.h>
@@ -52,6 +53,7 @@
 
 #include <Core/Streaming/CursorTree_fwd.h>
 
+#include <functional>
 #include <ranges>
 
 namespace DB
@@ -1168,24 +1170,82 @@ bool hasUnknownColumn(const QueryTreeNodePtr & node, QueryTreeNodePtr table_expr
     return false;
 }
 
-void removeExpressionsThatDoNotDependOnTableIdentifiers(
+namespace
+{
+
+template <typename KeepFunction>
+bool walkOrdinaryFunctions(const QueryTreeNodePtr & node, KeepFunction && keep_function)
+{
+    QueryTreeNodes stack = {node};
+    while (!stack.empty())
+    {
+        auto current = std::move(stack.back());
+        stack.pop_back();
+        if (!current)
+            continue;
+
+        const auto type = current->getNodeType();
+        if (type == QueryTreeNodeType::QUERY || type == QueryTreeNodeType::UNION)
+            return false;
+
+        if (const auto * function = current->as<FunctionNode>())
+        {
+            if (function->isWindowFunction() || function->isAggregateFunction())
+                return false;
+            if (function->isOrdinaryFunction())
+            {
+                auto function_base = function->getFunction();
+                if (!function_base || !keep_function(function_base))
+                    return false;
+            }
+        }
+
+        for (const auto & child : current->getChildren())
+        {
+            if (child)
+                stack.push_back(child);
+        }
+    }
+    return true;
+}
+
+bool isSafeToDuplicateInQueryTree(const QueryTreeNodePtr & node)
+{
+    return walkOrdinaryFunctions(
+        node,
+        [](const FunctionBasePtr & function_base)
+        {
+            return function_base->isDeterministic()
+                && function_base->isDeterministicInScopeOfQuery()
+                && !function_base->isStateful()
+                && !function_base->isServerConstant()
+                && !functionIsDictGet(function_base->getName())
+                && !functionIsJoinGet(function_base->getName());
+        });
+}
+
+void filterConjunctions(
     QueryTreeNodePtr & expression,
-    const QueryTreeNodePtr & table_expression,
+    const std::function<bool(const QueryTreeNodePtr &)> & keep,
     const ContextPtr & context)
 {
     auto * function = expression->as<FunctionNode>();
     if (!function)
+    {
+        if (!keep(expression))
+            expression = {};
         return;
+    }
 
     if (function->getFunctionName() != "and")
     {
-        if (hasUnknownColumn(expression, table_expression))
-            expression = nullptr;
+        if (!keep(expression))
+            expression = {};
         return;
     }
 
     QueryTreeNodesDeque conjunctions;
-    QueryTreeNodesDeque processing{ expression };
+    QueryTreeNodesDeque processing{expression};
 
     while (!processing.empty())
     {
@@ -1195,10 +1255,7 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
         if (auto * function_node = node->as<FunctionNode>())
         {
             if (function_node->getFunctionName() == "and")
-                std::ranges::copy(
-                    function_node->getArguments(),
-                    std::back_inserter(processing)
-                );
+                std::ranges::copy(function_node->getArguments(), std::back_inserter(processing));
             else
                 conjunctions.push_back(node);
         }
@@ -1212,7 +1269,7 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
 
     for (const auto & node : processing)
     {
-        if (!hasUnknownColumn(node, table_expression))
+        if (keep(node))
             conjunctions.push_back(node);
     }
 
@@ -1234,6 +1291,29 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
     function->resolveAsFunction(function_impl->build(function->getArgumentColumns()));
 }
 
+}
+
+void removeExpressionsThatDoNotDependOnTableIdentifiers(
+    QueryTreeNodePtr & expression,
+    const QueryTreeNodePtr & table_expression,
+    const ContextPtr & context)
+{
+    filterConjunctions(
+        expression,
+        [&](const QueryTreeNodePtr & node) { return !hasUnknownColumn(node, table_expression); },
+        context);
+}
+
+void removeExpressionsThatAreUnsafeToDuplicate(
+    QueryTreeNodePtr & expression,
+    const ContextPtr & context)
+{
+    if (!expression)
+        return;
+
+    filterConjunctions(expression, isSafeToDuplicateInQueryTree, context);
+}
+
 namespace
 {
 

src/Analyzer/Utils.h

@@ -208,13 +208,27 @@ bool hasUnknownColumn(
 /** Suppose we have a table x with columns a, c, d and
   * a an expression like x.a > 2 AND y.b > 3 AND x.c + 1 == x.d
   * This method will remove the part y.b > 3 from it since it depends
-  * on unknown columns from a different table.
+  * on unknown columns from a different table. A non-function root such as
+  * `WHERE y.b` is dropped the same way.
   */
 void removeExpressionsThatDoNotDependOnTableIdentifiers(
     QueryTreeNodePtr & expression,
     const QueryTreeNodePtr & replacement_table_expression,
     const ContextPtr & context);
 
+/** Remove conjuncts that are unsafe to copy into another query tree (not deterministic, not
+  * deterministic in this query, stateful, or server-constant). Nested `and` is flattened the same
+  * way as `removeExpressionsThatDoNotDependOnTableIdentifiers`. Window and aggregate functions are
+  * also dropped. JOIN filter pushdown refuses stateful predicates via
+  * `ActionsDAG::hasStatefulFunctions`.
+  *
+  * The wrap `WHERE` is sent to remote cluster nodes. Node-local functions such as `hostName`,
+  * `dictGet`, `joinGet`, `FQDN`, and `queryID` must stay on the initiator: remotes can miss the
+  * dictionary, see different data, or return a different server-local value.
+  */
+void removeExpressionsThatAreUnsafeToDuplicate(
+    QueryTreeNodePtr & expression,
+    const ContextPtr & context);
 
 Field getFieldFromColumnForASTLiteral(const ColumnPtr & column, size_t row, const DataTypePtr & data_type);
 

src/Core/Joins.h

@@ -140,6 +140,25 @@ enum class JoinTableSide : uint8_t
 
 const char * toString(JoinTableSide join_table_side);
 
+/** Whether ordinary columns from this side of a JOIN can be used as filter inputs
+  * before the JOIN. Skip the null-producing side of an outer JOIN, the right side
+  * of an `ASOF JOIN`, and both sides of a `PASTE JOIN` or `FULL JOIN`.
+  * Attaching an equivalent-key filter to the other child, and dictionary / lookup
+  * fill, are separate (`JoinStep::allowPushDownToRight`).
+  */
+constexpr bool canPrefilterJoinSide(JoinKind kind, JoinStrictness strictness, JoinTableSide side)
+{
+    if (isPaste(kind) || isFull(kind))
+        return false;
+    if (strictness == JoinStrictness::Asof && side == JoinTableSide::Right)
+        return false;
+    if (isLeft(kind) && side == JoinTableSide::Right)
+        return false;
+    if (isRight(kind) && side == JoinTableSide::Left)
+        return false;
+    return true;
+}
+
 enum class JoinOrderAlgorithm : uint8_t
 {
     GREEDY = 0,

src/Functions/FunctionJoinGet.cpp

@@ -82,6 +82,8 @@ class FunctionJoinGet final : public IFunctionBase
 
     String getName() const override { return function_name; }
 
+    bool isDeterministic() const override { return false; }
+
     bool isSuitableForShortCircuitArgumentsExecution(const DataTypesWithConstInfo & /*arguments*/) const override { return true; }
 
     const DataTypes & getArgumentTypes() const override { return argument_types; }

src/Planner/Planner.cpp

@@ -221,6 +221,11 @@ void checkStoragesSupportTransactions(const PlannerContextPtr & planner_context)
     }
 }
 
+}
+
+namespace
+{
+
 /** Storages can rely that filters that for storage will be available for analysis before
   * getQueryProcessingStage method will be called.
   *
@@ -390,6 +395,8 @@ FiltersForTableExpressionMap collectFiltersForAnalysis(const QueryTreeNodePtr &
     return res;
 }
 
+}
+
 FiltersForTableExpressionMap collectFiltersForAnalysis(const QueryTreeNodePtr & query_tree_node, const SelectQueryOptions & select_query_options, const ActionsDAG * post_filter)
 {
     if (select_query_options.only_analyze)
@@ -411,6 +418,9 @@ FiltersForTableExpressionMap collectFiltersForAnalysis(const QueryTreeNodePtr &
     return collectFiltersForAnalysis(query_tree_node, table_expressions_nodes, context, post_filter);
 }
 
+namespace
+{
+
 /// Extend lifetime of query context, storages, and table locks
 void extendQueryContextAndStoragesLifetime(QueryPlan & query_plan, const PlannerContextPtr & planner_context)
 {

src/Planner/Planner.h

@@ -7,6 +7,7 @@
 
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <Storages/SelectQueryInfo.h>
+#include <Planner/PlannerContext.h>
 
 namespace DB
 {
@@ -89,4 +90,9 @@ class Planner
     QueryNodeToPlanStepMapping query_node_to_plan_step_mapping;
 };
 
+FiltersForTableExpressionMap collectFiltersForAnalysis(
+    const QueryTreeNodePtr & query_tree_node,
+    const SelectQueryOptions & select_query_options,
+    const ActionsDAG * post_filter);
+
 }