Fix join filter pushdown

src/Analyzer/Utils.cpp

@@ -52,6 +52,7 @@
 
 #include <Core/Streaming/CursorTree_fwd.h>
 
+#include <functional>
 #include <ranges>
 
 namespace DB
@@ -1168,9 +1169,47 @@ bool hasUnknownColumn(const QueryTreeNodePtr & node, QueryTreeNodePtr table_expr
     return false;
 }
 
-void removeExpressionsThatDoNotDependOnTableIdentifiers(
+namespace
+{
+
+bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node)
+{
+    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 || !function_base->isDeterministicInScopeOfQuery())
+                    return false;
+            }
+        }
+
+        for (const auto & child : current->getChildren())
+        {
+            if (child)
+                stack.push_back(child);
+        }
+    }
+    return true;
+}
+
+void filterConjunctions(
     QueryTreeNodePtr & expression,
-    const QueryTreeNodePtr & table_expression,
+    const std::function<bool(const QueryTreeNodePtr &)> & keep,
     const ContextPtr & context)
 {
     auto * function = expression->as<FunctionNode>();
@@ -1179,13 +1218,13 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
 
     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 +1234,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 +1248,7 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
 
     for (const auto & node : processing)
     {
-        if (!hasUnknownColumn(node, table_expression))
+        if (keep(node))
             conjunctions.push_back(node);
     }
 
@@ -1234,6 +1270,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 removeExpressionsThatAreNotDeterministicInScopeOfQuery(
+    QueryTreeNodePtr & expression,
+    const ContextPtr & context)
+{
+    if (!expression)
+        return;
+
+    filterConjunctions(expression, isDeterministicInScopeOfQueryTree, context);
+}
+
 namespace
 {
 

src/Analyzer/Utils.h

@@ -215,6 +215,12 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers(
     const QueryTreeNodePtr & replacement_table_expression,
     const ContextPtr & context);
 
+/** Remove conjuncts that are not deterministic in the current query (`rand`, and similar).
+  * Nested `and` is flattened the same way as `removeExpressionsThatDoNotDependOnTableIdentifiers`.
+  */
+void removeExpressionsThatAreNotDeterministicInScopeOfQuery(
+    QueryTreeNodePtr & expression,
+    const ContextPtr & context);
 
 Field getFieldFromColumnForASTLiteral(const ColumnPtr & column, size_t row, const DataTypePtr & data_type);
 

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

src/Planner/PlannerJoinTree.cpp

@@ -66,6 +66,7 @@
 #include <Processors/QueryPlan/ReadFromTableStep.h>
 #include <Processors/QueryPlan/ReadFromTableFunctionStep.h>
 #include <Processors/QueryPlan/ReadNothingStep.h>
+#include <Processors/QueryPlan/SourceStepWithFilter.h>
 #include <Processors/QueryPlan/Optimizations/Utils.h>
 #include <Processors/QueryPlan/ParallelReplicasLocalPlan.h>
 #include <Processors/Sources/SourceFromSingleChunk.h>
@@ -215,6 +216,83 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const
     }
 }
 
+/// Same outer-join sides as JOIN filter pushdown / `FunctionToSubcolumnsPass`:
+/// do not copy a predicate onto the null-producing side.
+bool joinTreePreservesRowsForTable(const QueryTreeNodePtr & join_tree, const QueryTreeNodePtr & table)
+{
+    std::vector<QueryTreeNodePtr> stack = {join_tree};
+    while (!stack.empty())
+    {
+        auto node = std::move(stack.back());
+        stack.pop_back();
+        if (!node)
+            continue;
+
+        if (const auto * join = node->as<JoinNode>())
+        {
+            if (isRightOrFull(join->getKind()) && extractTableExpressionsSet(join->getLeftTableExpression()).contains(table.get()))
+                return false;
+            if (isLeftOrFull(join->getKind()) && extractTableExpressionsSet(join->getRightTableExpression()).contains(table.get()))
+                return false;
+            stack.push_back(join->getLeftTableExpression());
+            stack.push_back(join->getRightTableExpression());
+        }
+        else if (const auto * array_join = node->as<ArrayJoinNode>())
+        {
+            stack.push_back(array_join->getTableExpression());
+        }
+        else if (const auto * cross_join = node->as<CrossJoinNode>())
+        {
+            for (const auto & expr : cross_join->getTableExpressions())
+                stack.push_back(expr);
+        }
+    }
+    return true;
+}
+
+/// `IStorageCluster` JOINs wrap the left table in a subquery. Attach dummy-analysis
+/// filters to the wrap source for listing only; do not add a FilterStep, which would
+/// drop unused columns from the wrap header.
+void tryAddClusterWrapFilter(QueryPlan & query_plan, const TableExpressionData & table_expression_data)
+{
+    const auto & filter_actions = table_expression_data.getFilterActions();
+    if (!filter_actions || !query_plan.isInitialized())
+        return;
+
+    QueryPlan::Node * node = query_plan.getRootNode();
+    while (node && !node->children.empty())
+        node = node->children.front();
+
+    auto * source = node ? dynamic_cast<SourceStepWithFilter *>(node->step.get()) : nullptr;
+    if (!source)
+        return;
+
+    auto filter_dag = filter_actions->clone();
+    const auto filter_column_name = filter_dag.getOutputs().at(0)->result_name;
+    const auto & header = source->getOutputHeader();
+    ActionsDAG rename_dag(header->getColumnsWithTypeAndName());
+    const auto & identifier_to_name = table_expression_data.getColumnIdentifierToColumnName();
+
+    for (const auto * input : filter_dag.getInputs())
+    {
+        if (header->has(input->result_name))
+            continue;
+
+        auto it = identifier_to_name.find(input->result_name);
+        if (it == identifier_to_name.end() || !header->has(it->second))
+            continue;
+
+        const auto & physical = rename_dag.findInOutputs(it->second);
+        rename_dag.addOrReplaceInOutputs(rename_dag.addAlias(physical, input->result_name));
+    }
+
+    filter_dag = ActionsDAG::merge(std::move(rename_dag), std::move(filter_dag));
+    source->addFilter(std::move(filter_dag), filter_column_name);
+    /// Wrap subquery planning already called `applyFilters` with no predicate.
+    /// Apply now so icebergCluster listing is recreated with the WHERE.
+    source->SourceStepWithFilterBase::applyFilters();
+}
+
 bool shouldIgnoreQuotaAndLimits(const TableNode & table_node)
 {
     const auto & storage_id = table_node.getStorageID();
@@ -920,8 +998,49 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
 
     if (wrap_read_columns_in_subquery)
     {
+        auto original_table_expression = table_expression;
+
+        /// Subqueries inherit the outer GlobalPlannerContext, whose filter map is keyed by
+        /// outer table nodes. Collect filters for this JOIN query so icebergCluster listing
+        /// still sees left-only WHERE after the wrap.
+        if (!table_expression_data.getFilterActions() && select_query_info.query_tree)
+        {
+            auto collected = collectFiltersForAnalysis(select_query_info.query_tree, select_query_options, nullptr);
+            auto it = collected.find(table_expression);
+            if (it != collected.end() && it->second.filter_actions)
+                table_expression_data.setFilterActions(it->second.filter_actions->clone());
+        }
+
         auto columns = table_expression_data.getColumns();
-        table_expression = buildSubqueryToReadColumnsFromTableExpression(columns, table_expression, query_context);
+        table_expression = buildSubqueryToReadColumnsFromTableExpression(columns, original_table_expression, query_context);
+
+        /// Wrap is planned as `SELECT cols FROM icebergCluster` with no JOIN. Copy left-only
+        /// WHERE/PREWHERE so initiator file listing sees the same predicate as a single-table
+        /// `icebergCluster` read. Same helper as `IStorageCluster::updateQueryWithJoinToSendIfNeeded`.
+        /// Skip the null-producing side of an outer JOIN (`WHERE isNull(r.x)` on a `LEFT JOIN`).
+        if (const auto * parent_query = select_query_info.query_tree->as<QueryNode>();
+            parent_query && joinTreePreservesRowsForTable(parent_query->getJoinTree(), original_table_expression))
+        {
+            auto copy_left_only = [&](const QueryTreeNodePtr & predicate) -> QueryTreeNodePtr
+            {
+                auto cloned = predicate->clone();
+                removeExpressionsThatDoNotDependOnTableIdentifiers(cloned, original_table_expression, query_context);
+                removeExpressionsThatAreNotDeterministicInScopeOfQuery(cloned, query_context);
+                return cloned;
+            };
+
+            auto & wrap_query = table_expression->as<QueryNode &>();
+            if (parent_query->hasWhere())
+            {
+                if (auto pred = copy_left_only(parent_query->getWhere()))
+                    wrap_query.getWhere() = std::move(pred);
+            }
+            if (parent_query->hasPrewhere())
+            {
+                if (auto pred = copy_left_only(parent_query->getPrewhere()))
+                    wrap_query.getPrewhere() = std::move(pred);
+            }
+        }
     }
 
     auto * table_node = table_expression->as<TableNode>();
@@ -1491,19 +1610,24 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
         else
         {
             std::shared_ptr<GlobalPlannerContext> subquery_planner_context;
+            auto subquery_options = select_query_options.subquery();
             if (wrap_read_columns_in_subquery)
-                subquery_planner_context = std::make_shared<GlobalPlannerContext>(nullptr, nullptr, nullptr, FiltersForTableExpressionMap{});
+            {
+                subquery_planner_context = std::make_shared<GlobalPlannerContext>(
+                    nullptr, nullptr, nullptr, collectFiltersForAnalysis(table_expression, subquery_options, nullptr));
+            }
             else
                 subquery_planner_context = planner_context->getGlobalPlannerContext();
 
-            auto subquery_options = select_query_options.subquery();
             Planner subquery_planner(table_expression, subquery_options, subquery_planner_context);
             /// Propagate storage limits to subquery
             subquery_planner.addStorageLimits(*select_query_info.storage_limits);
             subquery_planner.buildQueryPlanIfNeeded();
             const auto & mapping = subquery_planner.getQueryNodeToPlanStepMapping();
             query_node_to_plan_step_mapping.insert(mapping.begin(), mapping.end());
             query_plan = std::move(subquery_planner).extractQueryPlan();
+            if (wrap_read_columns_in_subquery && till_stage == QueryProcessingStage::FetchColumns)
+                tryAddClusterWrapFilter(query_plan, table_expression_data);
         }
 
         auto & alias_column_expressions = table_expression_data.getAliasColumnExpressions();