Fix join filter pushdown

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,49 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const
     }
 }
 
+/// `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 +964,46 @@ 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`.
+        if (const auto * parent_query = select_query_info.query_tree->as<QueryNode>())
+        {
+            auto copy_left_only = [&](const QueryTreeNodePtr & predicate) -> QueryTreeNodePtr
+            {
+                auto cloned = predicate->clone();
+                removeExpressionsThatDoNotDependOnTableIdentifiers(cloned, original_table_expression, 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 +1573,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();

src/Processors/QueryPlan/Optimizations/filterPushDown.cpp

@@ -560,6 +560,10 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan::
         equivalent_expressions.append_range(std::move(extra_equivalent_expressions));
     }
 
+    NameSet filter_input_names;
+    for (const auto * input_node : filter->getExpression().getInputs())
+        filter_input_names.emplace(input_node->result_name);
+
     auto get_available_columns_for_filter = [&](bool push_to_left_stream, bool filter_push_down_input_columns_available, bool require_stable_types = false)
     {
         Names available_input_columns_for_filter;
@@ -568,11 +572,24 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan::
             return available_input_columns_for_filter;
 
         const auto & input_header = push_to_left_stream ? left_stream_input_header : right_stream_input_header;
-        const auto & input_columns_names = input_header->getNames();
+        NameSet already_added;
 
-        for (const auto & name : input_columns_names)
+        auto try_add = [&](const String & name)
         {
-            if (!join_header->has(name))
+            if (!already_added.insert(name).second)
+                return;
+
+            available_input_columns_for_filter.push_back(name);
+        };
+
+        for (const auto & name : input_header->getNames())
+        {
+            const bool in_join_output = join_header->has(name);
+
+            /// JOIN output may drop a left-only column (unused-column removal after
+            /// `count()` of `SELECT * … JOIN … WHERE left.col …`) while the Filter DAG
+            /// still references it. That name is still valid on this stream.
+            if (!in_join_output && (require_stable_types || !filter_input_names.contains(name)))
                 continue;
 
             /// For the legacy JoinStep (not JoinStepLogical), there is no mechanism to adjust
@@ -583,11 +600,44 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan::
             ///
             /// The disjunction (partial predicate) push-down path has no such type-fixup, so it
             /// passes require_stable_types to also exclude type-changing columns for JoinStepLogical.
-            if ((!logical_join || require_stable_types)
+            if (in_join_output
+                && (!logical_join || require_stable_types)
                 && !input_header->getByName(name).type->equals(*join_header->getByName(name).type))
                 continue;
 
-            available_input_columns_for_filter.push_back(name);
+            try_add(name);
+        }
+
+        /// JoinStepLogical may alias a side's input (`bid`) to a JOIN-output / filter name
+        /// (`__table1.bid`). `splitActionsForJOINFilterPushDown` matches filter inputs, so
+        /// the output name must be listed; `fix_predicate_for_join_logical_step` remaps it.
+        if (logical_join)
+        {
+            for (const auto & output_action : logical_join->getOutputActions())
+            {
+                if (push_to_left_stream ? !output_action.fromLeft() : !output_action.fromRight())
+                    continue;
+
+                const auto & output_name = output_action.getColumnName();
+                if (!join_header->has(output_name) && !filter_input_names.contains(output_name))
+                    continue;
+
+                if (require_stable_types)
+                {
+                    auto resolved = output_action.resolveAliases();
+                    if (resolved.getNode()->type != ActionsDAG::ActionType::INPUT
+                        || !input_header->has(resolved.getColumnName()))
+                        continue;
+
+                    const auto & output_type = join_header->has(output_name)
+                        ? join_header->getByName(output_name).type
+                        : output_action.getType();
+                    if (!input_header->getByName(resolved.getColumnName()).type->equals(*output_type))
+                        continue;
+                }
+
+                try_add(output_name);
+            }
         }
 
         return available_input_columns_for_filter;

src/Storages/IStorageCluster.cpp

@@ -98,7 +98,9 @@ void ReadFromCluster::applyFilters(ActionDAGNodes added_filter_nodes)
 
 void ReadFromCluster::createExtension(const ActionsDAG::Node * predicate)
 {
-    if (extension)
+    /// Listing is one-shot. Recreate only when a real predicate arrives after an
+    /// empty listing (e.g. `initializePipeline` ran before `applyFilters`).
+    if (extension && !(predicate && !extension_has_predicate))
         return;
 
     extension = storage->getTaskIteratorExtension(
@@ -107,6 +109,7 @@ void ReadFromCluster::createExtension(const ActionsDAG::Node * predicate)
         context,
         cluster,
         getStorageSnapshot()->metadata);
+    extension_has_predicate = predicate != nullptr;
 }
 
 namespace
@@ -596,7 +599,9 @@ void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const
     if (current_settings[Setting::max_parallel_replicas] > 1)
         max_replicas_to_use = std::min(max_replicas_to_use, current_settings[Setting::max_parallel_replicas].value);
 
-    createExtension(nullptr);
+    const ActionsDAG * filter = filter_actions_dag ? filter_actions_dag.get() : query_info.filter_actions_dag.get();
+    const ActionsDAG::Node * predicate = filter ? filter->getOutputs().at(0) : nullptr;
+    createExtension(predicate);
 
     ProfileEvents::increment(ProfileEvents::Shards, max_replicas_to_use);
 

src/Storages/IStorageCluster.h

@@ -172,6 +172,7 @@ class ReadFromCluster : public SourceStepWithFilter
     LoggerPtr log;
 
     std::optional<RemoteQueryExecutor::Extension> extension;
+    bool extension_has_predicate = false;
     std::optional<Tables> external_tables;
 
     void createExtension(const ActionsDAG::Node * predicate);