From 6f91aa6597d29f4ad1dbced20863a7334bc50e13 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 21 Aug 2026 17:52:45 +0200 Subject: [PATCH 1/3] Allow JOIN filter pushdown when side column names do not match the join header Unused-column removal and `JoinStepLogical` aliases can hide a one-sided `WHERE` from `get_available_columns_for_filter`. Include those names so existing split and remap can push the predicate under the JOIN. Co-authored-by: Cursor --- .../Optimizations/filterPushDown.cpp | 60 +++++++++++-- ...n_filter_pushdown_count_subquery.reference | 2 + ...73_join_filter_pushdown_count_subquery.sql | 86 +++++++++++++++++++ 3 files changed, 143 insertions(+), 5 deletions(-) create mode 100644 tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.reference create mode 100644 tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql diff --git a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp index 98ba6cc62cad..3e86fb520669 100644 --- a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp +++ b/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; diff --git a/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.reference b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.reference new file mode 100644 index 000000000000..9b231627ac1d --- /dev/null +++ b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.reference @@ -0,0 +1,2 @@ +40 +40 diff --git a/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql new file mode 100644 index 000000000000..58285f7d4da5 --- /dev/null +++ b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql @@ -0,0 +1,86 @@ +-- Tags: no-parallel-replicas +-- Left-only WHERE on `count()` of `SELECT * … JOIN` must still be pushed through +-- the JOIN (and composed through identifier-rename expressions) so the left +-- read can apply PREWHERE / index analysis. + +DROP TABLE IF EXISTS t_left; +DROP TABLE IF EXISTS t_right; + +CREATE TABLE t_left +( + a Int32, + b Int32 +) +ENGINE = MergeTree +ORDER BY a +SETTINGS index_granularity = 1024, index_granularity_bytes = '10Mi'; + +CREATE TABLE t_right +( + a Int32, + b Int32 +) +ENGINE = Memory; + +INSERT INTO t_left SELECT number, number FROM numbers(100); +INSERT INTO t_right SELECT number, number FROM numbers(100); + +SET enable_parallel_replicas = 0; +SET query_plan_join_swap_table = 0; +SET enable_analyzer = 1; +SET query_plan_filter_push_down = 1; +SET enable_join_runtime_filters = 0; +SET join_use_nulls = 1; + +SELECT count() +FROM +( + SELECT * + FROM t_left AS foo + LEFT JOIN t_right AS bar ON foo.b = bar.b + WHERE foo.a < 40 +); + +SELECT throwIf(count() = 0) +FROM +( + EXPLAIN actions = 1 + SELECT count() + FROM + ( + SELECT * + FROM t_left AS foo + LEFT JOIN t_right AS bar ON foo.b = bar.b + WHERE foo.a < 40 + ) +) +WHERE explain ILIKE '%Prewhere%' +FORMAT Null; + +SELECT count() +FROM +( + SELECT * + FROM (SELECT * FROM t_left) AS foo + LEFT JOIN t_right AS bar ON foo.b = bar.b + WHERE foo.a < 40 +); + +SELECT throwIf(count() = 0) +FROM +( + EXPLAIN actions = 1 + SELECT count() + FROM + ( + SELECT * + FROM (SELECT * FROM t_left) AS foo + LEFT JOIN t_right AS bar ON foo.b = bar.b + WHERE foo.a < 40 + ) +) +WHERE explain ILIKE '%Prewhere%' +FORMAT Null; + +DROP TABLE t_left; +DROP TABLE t_right; From af87b909d739710c457a1cd52d4d828ae8691b32 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 21 Aug 2026 19:26:16 +0200 Subject: [PATCH 2/3] Copy left-only WHERE into IStorageCluster JOIN wraps so icebergCluster can prune files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Initiator listing runs on the wrap subquery (`SELECT cols FROM icebergCluster`), which previously had no WHERE. A left-only predicate on `count()` of `SELECT * … JOIN` never reached min/max file listing. Co-authored-by: Cursor --- src/Planner/Planner.cpp | 10 ++ src/Planner/Planner.h | 6 + src/Planner/PlannerJoinTree.cpp | 102 +++++++++++++- .../optimizePrimaryKeyConditionAndLimit.cpp | 131 +++++++++++++++++- src/Storages/IStorageCluster.cpp | 9 +- src/Storages/IStorageCluster.h | 1 + ...test_cluster_join_filter_minmax_pruning.py | 129 +++++++++++++++++ 7 files changed, 377 insertions(+), 11 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_with_spark/test_cluster_join_filter_minmax_pruning.py diff --git a/src/Planner/Planner.cpp b/src/Planner/Planner.cpp index b36bc3b00cd7..faf2e42ea909 100644 --- a/src/Planner/Planner.cpp +++ b/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) { diff --git a/src/Planner/Planner.h b/src/Planner/Planner.h index 7e1c87d5f41f..7b6d7a35c80b 100644 --- a/src/Planner/Planner.h +++ b/src/Planner/Planner.h @@ -7,6 +7,7 @@ #include #include +#include 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); + } diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index dd4ef0a462a1..0d9e70dc4e6d 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -66,6 +66,7 @@ #include #include #include +#include #include #include #include @@ -215,6 +216,74 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const } } +bool whereOnlyReferencesTable(const QueryTreeNodePtr & where, const QueryTreeNodePtr & table) +{ + std::vector stack = {where}; + while (!stack.empty()) + { + auto current = std::move(stack.back()); + stack.pop_back(); + + if (const auto * column = current->as()) + { + auto source = column->getColumnSourceOrNull(); + if (!source || source.get() != table.get()) + return false; + } + + for (const auto & child : current->getChildren()) + { + if (child) + stack.push_back(child); + } + } + return true; +} + +/// `IStorageCluster` JOINs wrap the left table in a subquery planned with an empty +/// `FiltersForTableExpressionMap`, so initiator file listing would miss left-only WHERE. +/// 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(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 +989,30 @@ 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 a left-only + /// WHERE onto that subquery so initiator file listing sees the same predicate as a + /// single-table `icebergCluster` read (which already prunes). + if (const auto * parent_query = select_query_info.query_tree->as()) + { + if (parent_query->hasWhere() && whereOnlyReferencesTable(parent_query->getWhere(), original_table_expression)) + table_expression->as().getWhere() = parent_query->getWhere()->clone(); + } } auto * table_node = table_expression->as(); @@ -1491,12 +1582,15 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres else { std::shared_ptr subquery_planner_context; + auto subquery_options = select_query_options.subquery(); if (wrap_read_columns_in_subquery) - subquery_planner_context = std::make_shared(nullptr, nullptr, nullptr, FiltersForTableExpressionMap{}); + { + subquery_planner_context = std::make_shared( + 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); @@ -1504,6 +1598,8 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres 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(); diff --git a/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp b/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp index ef3608c98a99..799f966ed53f 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp @@ -1,13 +1,107 @@ #include #include #include +#include +#include #include #include #include +#include +#include +#include +#include + +#include namespace DB::QueryPlanOptimizations { +namespace +{ + +bool isJoinThatAcceptsLeftFilter(IQueryPlanStep * step) +{ + if (const auto * logical_join = typeid_cast(step)) + { + const auto kind = logical_join->getJoinOperator().kind; + return isInnerOrLeft(kind) || isCrossOrComma(kind); + } + if (const auto * join_step = typeid_cast(step)) + { + const auto kind = join_step->getJoin()->getTableJoin().kind(); + return isInnerOrLeft(kind) || isCrossOrComma(kind); + } + return false; +} + +bool typesCompatibleForSourceFilter(const DataTypePtr & header_type, const DataTypePtr & dag_type) +{ + if (header_type->equals(*dag_type)) + return true; + return removeNullableOrLowCardinalityNullable(header_type)->equals(*removeNullableOrLowCardinalityNullable(dag_type)); +} + +std::optional tryPhysicalNameInHeader(const std::string & name, const Block & header) +{ + if (header.has(name)) + return name; + + const auto pos = name.rfind('.'); + if (pos == std::string::npos || pos + 1 >= name.size()) + return {}; + + std::string suffix = name.substr(pos + 1); + if (suffix.size() >= 2 && suffix.front() == '`' && suffix.back() == '`') + suffix = suffix.substr(1, suffix.size() - 2); + + if (header.has(suffix)) + return suffix; + return {}; +} + +ActionsDAG remapFilterInputsToHeader(ActionsDAG filter_dag, const Block & header) +{ + ActionsDAG rename_dag(header.getColumnsWithTypeAndName()); + bool need_merge = false; + + for (const auto * input : filter_dag.getInputs()) + { + if (header.has(input->result_name) && typesCompatibleForSourceFilter(header.getByName(input->result_name).type, input->result_type)) + continue; + + auto physical = tryPhysicalNameInHeader(input->result_name, header); + if (!physical) + continue; + + const auto & node = rename_dag.findInOutputs(*physical); + rename_dag.addOrReplaceInOutputs(rename_dag.addAlias(node, input->result_name)); + need_merge = true; + } + + if (!need_merge) + return filter_dag; + + auto merged = ActionsDAG::merge(std::move(rename_dag), std::move(filter_dag)); + merged.removeUnusedActions(); + return merged; +} + +bool filterInputsAreInHeader(const ActionsDAG & filter_dag, const Block & header) +{ + for (const auto * input : filter_dag.getInputs()) + { + auto physical = tryPhysicalNameInHeader(input->result_name, header); + if (!physical) + return false; + if (!typesCompatibleForSourceFilter(header.getByName(*physical).type, input->result_type)) + return false; + } + return true; +} + +} + + void optimizePrimaryKeyConditionAndLimit(const Stack & stack) { const auto & frame = stack.back(); @@ -32,10 +126,15 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) /// analysis when plan optimizations like mergeExpressions have not /// merged these steps into the filter. std::vector expression_dags; + const QueryPlan::Node * coming_from = frame.node; + const auto & source_header = *source_step_with_filter->getOutputHeader(); + bool added_filter = false; for (auto iter = stack.rbegin() + 1; iter != stack.rend(); ++iter) { - if (auto * filter_step = typeid_cast(iter->node->step.get())) + auto * step = iter->node->step.get(); + + if (auto * filter_step = typeid_cast(step)) { auto filter_dag = filter_step->getExpression().clone(); auto filter_column_name = filter_step->getFilterColumnName(); @@ -47,14 +146,22 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) for (auto it = expression_dags.rbegin(); it != expression_dags.rend(); ++it) filter_dag = ActionsDAG::merge((*it)->clone(), std::move(filter_dag)); - source_step_with_filter->addFilter(std::move(filter_dag), filter_column_name); + filter_dag = remapFilterInputsToHeader(std::move(filter_dag), source_header); + + /// A filter above JOIN may reference the other side. Skip those; left-only + /// predicates still apply to this source (needed for icebergCluster listing). + if (filterInputsAreInHeader(filter_dag, source_header)) + { + source_step_with_filter->addFilter(std::move(filter_dag), filter_column_name); + added_filter = true; + } } - else if (auto * limit_step = typeid_cast(iter->node->step.get())) + else if (auto * limit_step = typeid_cast(step)) { source_step_with_filter->setLimit(limit_step->getLimitForSorting()); break; } - else if (auto * expression_step = typeid_cast(iter->node->step.get())) + else if (auto * expression_step = typeid_cast(step)) { /// `arrayJoin` in an `ExpressionStep` above the source changes row cardinality. /// Propagating the outer `LIMIT` past such a step is unsound: the source would @@ -68,16 +175,28 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) if (expression_step->getExpression().hasArrayJoin()) break; expression_dags.push_back(&expression_step->getExpression()); - continue; } - else if (auto * object_filter_step = typeid_cast(iter->node->step.get())) + else if (auto * object_filter_step = typeid_cast(step)) { source_step_with_filter->addFilter(object_filter_step->getExpression().clone(), object_filter_step->getFilterColumnName()); + added_filter = true; + } + else if ( + !added_filter + && isJoinThatAcceptsLeftFilter(step) + && !iter->node->children.empty() + && iter->node->children.front() == coming_from) + { + /// `icebergCluster` lists files during `applyFilters`, which previously + /// stopped at JOIN. If the left-only WHERE is still above the JOIN, + /// keep walking so file listing can prune. } else { break; } + + coming_from = iter->node; } source_step_with_filter->applyFilters(); diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 3012c7bff735..4fec16100b6d 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/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); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 9613f9549562..e9714bf7f694 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -172,6 +172,7 @@ class ReadFromCluster : public SourceStepWithFilter LoggerPtr log; std::optional extension; + bool extension_has_predicate = false; std::optional external_tables; void createExtension(const ActionsDAG::Node * predicate); diff --git a/tests/integration/test_storage_iceberg_with_spark/test_cluster_join_filter_minmax_pruning.py b/tests/integration/test_storage_iceberg_with_spark/test_cluster_join_filter_minmax_pruning.py new file mode 100644 index 000000000000..fb1e5009d89a --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/test_cluster_join_filter_minmax_pruning.py @@ -0,0 +1,129 @@ +import pytest + +from helpers.iceberg_utils import ( + check_validity_and_get_prunned_files_general, + execute_spark_query_general, + get_creation_expression, + get_uuid_str, +) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_cluster_join_filter_minmax_pruning(started_cluster_iceberg_with_spark, storage_type): + """ + icebergCluster lists files on the initiator. A left-only WHERE on + count() of SELECT * … JOIN must still reach that listing so min/max + pruning can skip files (the original icebergCluster JOIN subquery case). + """ + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_cluster_join_filter_minmax_pruning_" + storage_type + "_" + get_uuid_str() + BAR_NAME = "bar_" + storage_type + "_" + get_uuid_str() + + def execute_spark_query(query: str): + return execute_spark_query_general( + spark, + started_cluster_iceberg_with_spark, + storage_type, + TABLE_NAME, + query, + ) + + execute_spark_query( + f""" + CREATE TABLE {TABLE_NAME} ( + datetime DATE, + symbol VARCHAR(50), + bid INT + ) + USING iceberg + OPTIONS('format-version'='2') + """ + ) + + execute_spark_query(f"INSERT INTO {TABLE_NAME} VALUES (DATE '2024-01-01', 'AAPL', 1)") + execute_spark_query(f"INSERT INTO {TABLE_NAME} VALUES (DATE '2024-01-02', 'AAPL', 2)") + execute_spark_query(f"INSERT INTO {TABLE_NAME} VALUES (DATE '2024-01-03', 'AAPL', 3)") + + iceberg = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + run_on_cluster=True, + ) + + instance.query( + f"CREATE TABLE `{BAR_NAME}` (symbol String, comment String) ENGINE = Memory" + ) + instance.query( + f"INSERT INTO `{BAR_NAME}` VALUES ('AAPL', 'comment'), ('AAPL2', 'comment2')" + ) + + common_settings = { + "input_format_parquet_bloom_filter_push_down": 0, + "input_format_parquet_filter_push_down": 0, + "query_plan_filter_push_down": 1, + "enable_analyzer": 1, + "query_plan_join_swap_table": 0, + "enable_join_runtime_filters": 0, + "enable_parallel_replicas": 0, + "join_use_nulls": 1, + } + + def check_validity_and_get_prunned_files(select_expression): + settings1 = {**common_settings, "use_iceberg_partition_pruning": 0} + settings2 = {**common_settings, "use_iceberg_partition_pruning": 1} + return check_validity_and_get_prunned_files_general( + instance, + TABLE_NAME, + settings1, + settings2, + "IcebergMinMaxIndexPrunedFiles", + select_expression, + ) + + # Three data files with disjoint bid ranges; bid >= 3 keeps one file. + expected_pruned = 2 + + assert ( + check_validity_and_get_prunned_files( + f"SELECT count() FROM {iceberg} WHERE bid >= 3" + ) + == expected_pruned + ) + + assert ( + check_validity_and_get_prunned_files( + f""" + SELECT count() + FROM {iceberg} AS foo + LEFT JOIN `{BAR_NAME}` AS bar ON foo.symbol = bar.symbol + WHERE foo.bid >= 3 + """ + ) + == expected_pruned + ) + + assert ( + check_validity_and_get_prunned_files( + f"SELECT count() FROM (SELECT * FROM {iceberg} AS foo WHERE foo.bid >= 3)" + ) + == expected_pruned + ) + + assert ( + check_validity_and_get_prunned_files( + f""" + SELECT count() + FROM + ( + SELECT * + FROM {iceberg} AS foo + LEFT JOIN `{BAR_NAME}` AS bar ON foo.symbol = bar.symbol + WHERE foo.bid >= 3 + ) + """ + ) + == expected_pruned + ) From 4e1b75445124ca28ada0b15f3f8783c3582e99d2 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 21 Aug 2026 19:47:20 +0200 Subject: [PATCH 3/3] Reuse existing left-only predicate helper for IStorageCluster JOIN wraps Drop the duplicated WHERE walker and the PK-walk-through-JOIN remapping. Wrap listing still uses collectFiltersForAnalysis and tryAddClusterWrapFilter. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 57 ++++---- .../optimizePrimaryKeyConditionAndLimit.cpp | 131 +----------------- 2 files changed, 30 insertions(+), 158 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 0d9e70dc4e6d..12e25b605fb4 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -216,34 +216,9 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const } } -bool whereOnlyReferencesTable(const QueryTreeNodePtr & where, const QueryTreeNodePtr & table) -{ - std::vector stack = {where}; - while (!stack.empty()) - { - auto current = std::move(stack.back()); - stack.pop_back(); - - if (const auto * column = current->as()) - { - auto source = column->getColumnSourceOrNull(); - if (!source || source.get() != table.get()) - return false; - } - - for (const auto & child : current->getChildren()) - { - if (child) - stack.push_back(child); - } - } - return true; -} - -/// `IStorageCluster` JOINs wrap the left table in a subquery planned with an empty -/// `FiltersForTableExpressionMap`, so initiator file listing would miss left-only WHERE. -/// 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. +/// `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(); @@ -1005,13 +980,29 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres auto columns = table_expression_data.getColumns(); table_expression = buildSubqueryToReadColumnsFromTableExpression(columns, original_table_expression, query_context); - /// Wrap is planned as `SELECT cols FROM icebergCluster` with no JOIN. Copy a left-only - /// WHERE onto that subquery so initiator file listing sees the same predicate as a - /// single-table `icebergCluster` read (which already prunes). + /// 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()) { - if (parent_query->hasWhere() && whereOnlyReferencesTable(parent_query->getWhere(), original_table_expression)) - table_expression->as().getWhere() = parent_query->getWhere()->clone(); + 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(); + 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); + } } } diff --git a/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp b/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp index 799f966ed53f..ef3608c98a99 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp @@ -1,107 +1,13 @@ #include #include #include -#include -#include #include #include #include -#include -#include -#include -#include - -#include namespace DB::QueryPlanOptimizations { -namespace -{ - -bool isJoinThatAcceptsLeftFilter(IQueryPlanStep * step) -{ - if (const auto * logical_join = typeid_cast(step)) - { - const auto kind = logical_join->getJoinOperator().kind; - return isInnerOrLeft(kind) || isCrossOrComma(kind); - } - if (const auto * join_step = typeid_cast(step)) - { - const auto kind = join_step->getJoin()->getTableJoin().kind(); - return isInnerOrLeft(kind) || isCrossOrComma(kind); - } - return false; -} - -bool typesCompatibleForSourceFilter(const DataTypePtr & header_type, const DataTypePtr & dag_type) -{ - if (header_type->equals(*dag_type)) - return true; - return removeNullableOrLowCardinalityNullable(header_type)->equals(*removeNullableOrLowCardinalityNullable(dag_type)); -} - -std::optional tryPhysicalNameInHeader(const std::string & name, const Block & header) -{ - if (header.has(name)) - return name; - - const auto pos = name.rfind('.'); - if (pos == std::string::npos || pos + 1 >= name.size()) - return {}; - - std::string suffix = name.substr(pos + 1); - if (suffix.size() >= 2 && suffix.front() == '`' && suffix.back() == '`') - suffix = suffix.substr(1, suffix.size() - 2); - - if (header.has(suffix)) - return suffix; - return {}; -} - -ActionsDAG remapFilterInputsToHeader(ActionsDAG filter_dag, const Block & header) -{ - ActionsDAG rename_dag(header.getColumnsWithTypeAndName()); - bool need_merge = false; - - for (const auto * input : filter_dag.getInputs()) - { - if (header.has(input->result_name) && typesCompatibleForSourceFilter(header.getByName(input->result_name).type, input->result_type)) - continue; - - auto physical = tryPhysicalNameInHeader(input->result_name, header); - if (!physical) - continue; - - const auto & node = rename_dag.findInOutputs(*physical); - rename_dag.addOrReplaceInOutputs(rename_dag.addAlias(node, input->result_name)); - need_merge = true; - } - - if (!need_merge) - return filter_dag; - - auto merged = ActionsDAG::merge(std::move(rename_dag), std::move(filter_dag)); - merged.removeUnusedActions(); - return merged; -} - -bool filterInputsAreInHeader(const ActionsDAG & filter_dag, const Block & header) -{ - for (const auto * input : filter_dag.getInputs()) - { - auto physical = tryPhysicalNameInHeader(input->result_name, header); - if (!physical) - return false; - if (!typesCompatibleForSourceFilter(header.getByName(*physical).type, input->result_type)) - return false; - } - return true; -} - -} - - void optimizePrimaryKeyConditionAndLimit(const Stack & stack) { const auto & frame = stack.back(); @@ -126,15 +32,10 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) /// analysis when plan optimizations like mergeExpressions have not /// merged these steps into the filter. std::vector expression_dags; - const QueryPlan::Node * coming_from = frame.node; - const auto & source_header = *source_step_with_filter->getOutputHeader(); - bool added_filter = false; for (auto iter = stack.rbegin() + 1; iter != stack.rend(); ++iter) { - auto * step = iter->node->step.get(); - - if (auto * filter_step = typeid_cast(step)) + if (auto * filter_step = typeid_cast(iter->node->step.get())) { auto filter_dag = filter_step->getExpression().clone(); auto filter_column_name = filter_step->getFilterColumnName(); @@ -146,22 +47,14 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) for (auto it = expression_dags.rbegin(); it != expression_dags.rend(); ++it) filter_dag = ActionsDAG::merge((*it)->clone(), std::move(filter_dag)); - filter_dag = remapFilterInputsToHeader(std::move(filter_dag), source_header); - - /// A filter above JOIN may reference the other side. Skip those; left-only - /// predicates still apply to this source (needed for icebergCluster listing). - if (filterInputsAreInHeader(filter_dag, source_header)) - { - source_step_with_filter->addFilter(std::move(filter_dag), filter_column_name); - added_filter = true; - } + source_step_with_filter->addFilter(std::move(filter_dag), filter_column_name); } - else if (auto * limit_step = typeid_cast(step)) + else if (auto * limit_step = typeid_cast(iter->node->step.get())) { source_step_with_filter->setLimit(limit_step->getLimitForSorting()); break; } - else if (auto * expression_step = typeid_cast(step)) + else if (auto * expression_step = typeid_cast(iter->node->step.get())) { /// `arrayJoin` in an `ExpressionStep` above the source changes row cardinality. /// Propagating the outer `LIMIT` past such a step is unsound: the source would @@ -175,28 +68,16 @@ void optimizePrimaryKeyConditionAndLimit(const Stack & stack) if (expression_step->getExpression().hasArrayJoin()) break; expression_dags.push_back(&expression_step->getExpression()); + continue; } - else if (auto * object_filter_step = typeid_cast(step)) + else if (auto * object_filter_step = typeid_cast(iter->node->step.get())) { source_step_with_filter->addFilter(object_filter_step->getExpression().clone(), object_filter_step->getFilterColumnName()); - added_filter = true; - } - else if ( - !added_filter - && isJoinThatAcceptsLeftFilter(step) - && !iter->node->children.empty() - && iter->node->children.front() == coming_from) - { - /// `icebergCluster` lists files during `applyFilters`, which previously - /// stopped at JOIN. If the left-only WHERE is still above the JOIN, - /// keep walking so file listing can prune. } else { break; } - - coming_from = iter->node; } source_step_with_filter->applyFilters();