From 6f91aa6597d29f4ad1dbced20863a7334bc50e13 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 21 Aug 2026 17:52:45 +0200 Subject: [PATCH 01/14] 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 02/14] 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 03/14] 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(); From a54a78aa155ff2898e0e300a0550fd58ab190880 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 11:44:21 +0200 Subject: [PATCH 04/14] Do not copy wrap predicates onto the null-producing side of an outer JOIN Copying a table-local WHERE such as `isNull(r.x)` under a LEFT JOIN remote right table changes join semantics. Use the same `isLeftOrFull` / `isRightOrFull` sides as JOIN filter pushdown. Related: https://github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 38 ++++++++++++++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 12e25b605fb4..5f51b943538e 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -216,6 +216,40 @@ 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 stack = {join_tree}; + while (!stack.empty()) + { + auto node = std::move(stack.back()); + stack.pop_back(); + if (!node) + continue; + + if (const auto * join = node->as()) + { + 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()) + { + stack.push_back(array_join->getTableExpression()); + } + else if (const auto * cross_join = node->as()) + { + 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. @@ -983,7 +1017,9 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres /// 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()) + /// 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(); + parent_query && joinTreePreservesRowsForTable(parent_query->getJoinTree(), original_table_expression)) { auto copy_left_only = [&](const QueryTreeNodePtr & predicate) -> QueryTreeNodePtr { From dc2f772de27b5de221b47c94191411c296c19e20 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 12:04:37 +0200 Subject: [PATCH 05/14] Do not copy nondeterministic wrap predicates that would run twice Share the existing `and`-conjunct filter in Analyzer/Utils so wrap copy can drop `rand` and similar after the table-local strip. Those conjuncts stay only in the original WHERE above the JOIN. Related: https://github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Analyzer/Utils.cpp | 79 ++++++++++++++++++++++++++++----- src/Analyzer/Utils.h | 6 +++ src/Planner/PlannerJoinTree.cpp | 1 + 3 files changed, 76 insertions(+), 10 deletions(-) diff --git a/src/Analyzer/Utils.cpp b/src/Analyzer/Utils.cpp index 939b8b3604ba..451bfbce9379 100644 --- a/src/Analyzer/Utils.cpp +++ b/src/Analyzer/Utils.cpp @@ -52,6 +52,7 @@ #include +#include #include 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()) + { + 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 & keep, const ContextPtr & context) { auto * function = expression->as(); @@ -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()) { 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 { diff --git a/src/Analyzer/Utils.h b/src/Analyzer/Utils.h index 9a19af2b4e0d..388356021fe1 100644 --- a/src/Analyzer/Utils.h +++ b/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); diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 5f51b943538e..a4902f40f249 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -1025,6 +1025,7 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres { auto cloned = predicate->clone(); removeExpressionsThatDoNotDependOnTableIdentifiers(cloned, original_table_expression, query_context); + removeExpressionsThatAreNotDeterministicInScopeOfQuery(cloned, query_context); return cloned; }; From 78a27b3767ca1a7fb4560cf2402189eb72dde174 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 12:13:52 +0200 Subject: [PATCH 06/14] Remove unnecessary `no-parallel-replicas` tag from JOIN filter pushdown test `SET enable_parallel_replicas = 0` already pins the EXPLAIN plan, so skipping the ParallelReplicas suite is not needed. Co-authored-by: Cursor --- .../0_stateless/04673_join_filter_pushdown_count_subquery.sql | 1 - 1 file changed, 1 deletion(-) 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 index 58285f7d4da5..f0e7ebab062c 100644 --- a/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql +++ b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql @@ -1,4 +1,3 @@ --- 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. From 28b3de8d1d81536a59eaa9bdfdb5ec88f997d107 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 14:35:31 +0200 Subject: [PATCH 07/14] Do not copy wrap predicates onto `ASOF` right or `PASTE` JOIN sides Prefiltering those sides changes nearest-match and positional pairing. Skip the same sides for wrap listing so icebergCluster cannot drop the matching file either. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 35 +++++++--- ...4_cluster_wrap_asof_paste_filter.reference | 2 + .../04674_cluster_wrap_asof_paste_filter.sql | 70 +++++++++++++++++++ 3 files changed, 97 insertions(+), 10 deletions(-) create mode 100644 tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.reference create mode 100644 tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.sql diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index a4902f40f249..ee839b6d3466 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -216,8 +216,10 @@ 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. +/// Same restrictions as JOIN filter pushdown (`filterPushDown.cpp`): +/// do not prefilter the null-producing side of an outer JOIN, the right side of +/// an `ASOF JOIN` (nearest-match would change), or either side of a `PASTE JOIN` +/// (positional alignment). bool joinTreePreservesRowsForTable(const QueryTreeNodePtr & join_tree, const QueryTreeNodePtr & table) { std::vector stack = {join_tree}; @@ -230,9 +232,16 @@ bool joinTreePreservesRowsForTable(const QueryTreeNodePtr & join_tree, const Que if (const auto * join = node->as()) { - if (isRightOrFull(join->getKind()) && extractTableExpressionsSet(join->getLeftTableExpression()).contains(table.get())) + const bool table_on_left = extractTableExpressionsSet(join->getLeftTableExpression()).contains(table.get()); + const bool table_on_right = extractTableExpressionsSet(join->getRightTableExpression()).contains(table.get()); + + if (isPaste(join->getKind()) && (table_on_left || table_on_right)) + return false; + if (join->getStrictness() == JoinStrictness::Asof && table_on_right) return false; - if (isLeftOrFull(join->getKind()) && extractTableExpressionsSet(join->getRightTableExpression()).contains(table.get())) + if (isRightOrFull(join->getKind()) && table_on_left) + return false; + if (isLeftOrFull(join->getKind()) && table_on_right) return false; stack.push_back(join->getLeftTableExpression()); stack.push_back(join->getRightTableExpression()); @@ -995,15 +1004,23 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres auto & table_expression_data = planner_context->getTableExpressionDataOrThrow(table_expression); QueryProcessingStage::Enum till_stage = QueryProcessingStage::Enum::FetchColumns; + bool can_prefilter_wrapped_table = false; if (wrap_read_columns_in_subquery) { auto original_table_expression = table_expression; + const auto * parent_query = select_query_info.query_tree + ? select_query_info.query_tree->as() + : nullptr; + can_prefilter_wrapped_table = parent_query + && joinTreePreservesRowsForTable(parent_query->getJoinTree(), original_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) + /// still sees left-only WHERE after the wrap. Skip the same join sides as + /// `joinTreePreservesRowsForTable` so listing cannot change `ASOF` / `PASTE` matches. + if (can_prefilter_wrapped_table && !table_expression_data.getFilterActions()) { auto collected = collectFiltersForAnalysis(select_query_info.query_tree, select_query_options, nullptr); auto it = collected.find(table_expression); @@ -1017,9 +1034,7 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres /// 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(); - parent_query && joinTreePreservesRowsForTable(parent_query->getJoinTree(), original_table_expression)) + if (can_prefilter_wrapped_table) { auto copy_left_only = [&](const QueryTreeNodePtr & predicate) -> QueryTreeNodePtr { @@ -1626,7 +1641,7 @@ 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) + if (wrap_read_columns_in_subquery && till_stage == QueryProcessingStage::FetchColumns && can_prefilter_wrapped_table) tryAddClusterWrapFilter(query_plan, table_expression_data); } diff --git a/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.reference b/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.reference new file mode 100644 index 000000000000..3c83ca38c2a7 --- /dev/null +++ b/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.reference @@ -0,0 +1,2 @@ +0 +1 b diff --git a/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.sql b/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.sql new file mode 100644 index 000000000000..d434660e4e46 --- /dev/null +++ b/tests/queries/0_stateless/04674_cluster_wrap_asof_paste_filter.sql @@ -0,0 +1,70 @@ +-- Tags: no-fasttest +-- no-fasttest: `fileCluster` is not in the fast test build. +-- Copying a wrap `WHERE` onto `IStorageCluster` must not prefilter the right +-- side of an `ASOF JOIN` or either side of a `PASTE JOIN`. + +SET enable_analyzer = 1; +SET query_plan_filter_push_down = 1; +SET query_plan_join_swap_table = 0; +SET enable_join_runtime_filters = 0; +SET enable_parallel_replicas = 0; + +DROP TABLE IF EXISTS t_asof_left; +CREATE TABLE t_asof_left +( + id Int32, + t Int32 +) +ENGINE = Memory; +INSERT INTO t_asof_left VALUES (1, 10); + +INSERT INTO FUNCTION file(currentDatabase() || '_04674_asof_right.tsv', 'TSV', 'id Int32, t Int32, flag Int32') +SELECT * +FROM +( + SELECT 1 AS id, 9 AS t, 0 AS flag + UNION ALL + SELECT 1, 8, 1 +) +SETTINGS engine_file_truncate_on_insert = 1; + +-- Nearest right row is `(t = 9, flag = 0)`. `WHERE flag = 1` must run after +-- `ASOF`, so the result is empty. Prefiltering the wrapped `fileCluster` would +-- keep only `t = 8` and incorrectly match it. +SELECT count() +FROM t_asof_left AS l +ASOF JOIN fileCluster( + 'test_cluster_one_shard_two_replicas', + currentDatabase() || '_04674_asof_right.tsv', + 'TSV', + 'id Int32, t Int32, flag Int32') AS r ON l.id = r.id AND l.t >= r.t +WHERE r.flag = 1; + +DROP TABLE t_asof_left; + +INSERT INTO FUNCTION file(currentDatabase() || '_04674_paste_left.tsv', 'TSV', 'n Int32, flag Int32') +SELECT number, if(number = 1, 1, 0) +FROM numbers(3) +SETTINGS engine_file_truncate_on_insert = 1; + +DROP TABLE IF EXISTS t_paste_right; +CREATE TABLE t_paste_right +( + s String +) +ENGINE = Memory; +INSERT INTO t_paste_right VALUES ('a'), ('b'), ('c'); + +-- `PASTE` pairs by position, then `WHERE flag = 1` keeps the middle pair +-- `(1, b)`. Prefiltering the wrapped left table would pair `1` with `a`. +SELECT l.n, r.s +FROM fileCluster( + 'test_cluster_one_shard_two_replicas', + currentDatabase() || '_04674_paste_left.tsv', + 'TSV', + 'n Int32, flag Int32') AS l +PASTE JOIN t_paste_right AS r +WHERE l.flag = 1 +SETTINGS max_threads = 1; + +DROP TABLE t_paste_right; From 0e141628c50e485a338df5106484dcdd26ba845e Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 14:51:24 +0200 Subject: [PATCH 08/14] Do not copy stateful wrap predicates that would run twice Functions such as `aiEmbed` and `timeSeriesStoreTags` can be deterministic in a query while still having side effects. Skip them in the cluster wrap the same way JOIN filter pushdown uses `hasStatefulFunctions`. Co-authored-by: Cursor --- src/Analyzer/Utils.cpp | 29 +++++++++++++-- src/Analyzer/Utils.h | 8 +++++ src/Planner/PlannerJoinTree.cpp | 1 + ...675_cluster_wrap_stateful_filter.reference | 1 + .../04675_cluster_wrap_stateful_filter.sql | 35 +++++++++++++++++++ 5 files changed, 72 insertions(+), 2 deletions(-) create mode 100644 tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.reference create mode 100644 tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.sql diff --git a/src/Analyzer/Utils.cpp b/src/Analyzer/Utils.cpp index 451bfbce9379..32511b97f339 100644 --- a/src/Analyzer/Utils.cpp +++ b/src/Analyzer/Utils.cpp @@ -1172,7 +1172,8 @@ bool hasUnknownColumn(const QueryTreeNodePtr & node, QueryTreeNodePtr table_expr namespace { -bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node) +template +bool walkOrdinaryFunctions(const QueryTreeNodePtr & node, KeepFunction && keep_function) { QueryTreeNodes stack = {node}; while (!stack.empty()) @@ -1193,7 +1194,7 @@ bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node) if (function->isOrdinaryFunction()) { auto function_base = function->getFunction(); - if (!function_base || !function_base->isDeterministicInScopeOfQuery()) + if (!function_base || !keep_function(function_base)) return false; } } @@ -1207,6 +1208,20 @@ bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node) return true; } +bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node) +{ + return walkOrdinaryFunctions( + node, + [](const FunctionBasePtr & function_base) { return function_base->isDeterministicInScopeOfQuery(); }); +} + +bool isStatelessInQueryTree(const QueryTreeNodePtr & node) +{ + return walkOrdinaryFunctions( + node, + [](const FunctionBasePtr & function_base) { return !function_base->isStateful(); }); +} + void filterConjunctions( QueryTreeNodePtr & expression, const std::function & keep, @@ -1293,6 +1308,16 @@ void removeExpressionsThatAreNotDeterministicInScopeOfQuery( filterConjunctions(expression, isDeterministicInScopeOfQueryTree, context); } +void removeExpressionsThatAreStateful( + QueryTreeNodePtr & expression, + const ContextPtr & context) +{ + if (!expression) + return; + + filterConjunctions(expression, isStatelessInQueryTree, context); +} + namespace { diff --git a/src/Analyzer/Utils.h b/src/Analyzer/Utils.h index 388356021fe1..7f4308385909 100644 --- a/src/Analyzer/Utils.h +++ b/src/Analyzer/Utils.h @@ -222,6 +222,14 @@ void removeExpressionsThatAreNotDeterministicInScopeOfQuery( QueryTreeNodePtr & expression, const ContextPtr & context); +/** Remove conjuncts that call stateful functions (`aiEmbed`, `timeSeriesStoreTags`, and similar). + * Those can report `isDeterministicInScopeOfQuery` while still having side effects; JOIN filter + * pushdown refuses them via `ActionsDAG::hasStatefulFunctions`. + */ +void removeExpressionsThatAreStateful( + QueryTreeNodePtr & expression, + const ContextPtr & context); + Field getFieldFromColumnForASTLiteral(const ColumnPtr & column, size_t row, const DataTypePtr & data_type); } diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index ee839b6d3466..6fb7e7f88bc6 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -1041,6 +1041,7 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres auto cloned = predicate->clone(); removeExpressionsThatDoNotDependOnTableIdentifiers(cloned, original_table_expression, query_context); removeExpressionsThatAreNotDeterministicInScopeOfQuery(cloned, query_context); + removeExpressionsThatAreStateful(cloned, query_context); return cloned; }; diff --git a/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.reference b/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.reference new file mode 100644 index 000000000000..0cfbf08886fc --- /dev/null +++ b/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.reference @@ -0,0 +1 @@ +2 diff --git a/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.sql b/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.sql new file mode 100644 index 000000000000..dc0709966b03 --- /dev/null +++ b/tests/queries/0_stateless/04675_cluster_wrap_stateful_filter.sql @@ -0,0 +1,35 @@ +-- Tags: no-fasttest +-- no-fasttest: `fileCluster` is not in the fast test build. +-- `timeSeriesStoreTags` is stateful and still `isDeterministicInScopeOfQuery`. +-- Copying it into the `IStorageCluster` wrap would store tags twice (the same +-- class of bug as wrapping `aiEmbed`). The `n < 2` conjunct must still be copied. + +SET enable_analyzer = 1; +SET query_plan_filter_push_down = 1; +SET query_plan_join_swap_table = 0; +SET enable_join_runtime_filters = 0; +SET enable_parallel_replicas = 0; + +INSERT INTO FUNCTION file(currentDatabase() || '_04675_wrap_left.tsv', 'TSV', 'n UInt64') +SELECT number +FROM numbers(3) +SETTINGS engine_file_truncate_on_insert = 1; + +DROP TABLE IF EXISTS t_wrap_right; +CREATE TABLE t_wrap_right +( + n UInt64 +) +ENGINE = Memory; +INSERT INTO t_wrap_right VALUES (0), (1), (2); + +SELECT count() +FROM fileCluster( + 'test_cluster_one_shard_two_replicas', + currentDatabase() || '_04675_wrap_left.tsv', + 'TSV', + 'n UInt64') AS l +LEFT JOIN t_wrap_right AS r ON l.n = r.n +WHERE l.n < 2 AND timeSeriesStoreTags(l.n, []) = l.n; + +DROP TABLE t_wrap_right; From 707dcdc3f2deabe8153df1803371b9a3d1e2de2a Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 15:57:09 +0200 Subject: [PATCH 09/14] Share JOIN prefilter side rules between wrap copy and filter pushdown Wrap copy restated the same outer/`ASOF`/`PASTE`/`FULL` checks as filter pushdown, and stripped wrap-unsafe conjuncts in two walks. Co-authored-by: Cursor --- src/Analyzer/Utils.cpp | 28 +++------- src/Analyzer/Utils.h | 16 ++---- src/Core/Joins.h | 18 ++++++ src/Planner/PlannerJoinTree.cpp | 16 ++---- .../Optimizations/filterPushDown.cpp | 56 ++++++++++--------- 5 files changed, 65 insertions(+), 69 deletions(-) diff --git a/src/Analyzer/Utils.cpp b/src/Analyzer/Utils.cpp index 32511b97f339..b539ae863ace 100644 --- a/src/Analyzer/Utils.cpp +++ b/src/Analyzer/Utils.cpp @@ -1208,18 +1208,14 @@ bool walkOrdinaryFunctions(const QueryTreeNodePtr & node, KeepFunction && keep_f return true; } -bool isDeterministicInScopeOfQueryTree(const QueryTreeNodePtr & node) +bool isSafeToDuplicateInQueryTree(const QueryTreeNodePtr & node) { return walkOrdinaryFunctions( node, - [](const FunctionBasePtr & function_base) { return function_base->isDeterministicInScopeOfQuery(); }); -} - -bool isStatelessInQueryTree(const QueryTreeNodePtr & node) -{ - return walkOrdinaryFunctions( - node, - [](const FunctionBasePtr & function_base) { return !function_base->isStateful(); }); + [](const FunctionBasePtr & function_base) + { + return function_base->isDeterministicInScopeOfQuery() && !function_base->isStateful(); + }); } void filterConjunctions( @@ -1298,24 +1294,14 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers( context); } -void removeExpressionsThatAreNotDeterministicInScopeOfQuery( - QueryTreeNodePtr & expression, - const ContextPtr & context) -{ - if (!expression) - return; - - filterConjunctions(expression, isDeterministicInScopeOfQueryTree, context); -} - -void removeExpressionsThatAreStateful( +void removeExpressionsThatAreUnsafeToDuplicate( QueryTreeNodePtr & expression, const ContextPtr & context) { if (!expression) return; - filterConjunctions(expression, isStatelessInQueryTree, context); + filterConjunctions(expression, isSafeToDuplicateInQueryTree, context); } namespace diff --git a/src/Analyzer/Utils.h b/src/Analyzer/Utils.h index 7f4308385909..b6ba1cd8f8f6 100644 --- a/src/Analyzer/Utils.h +++ b/src/Analyzer/Utils.h @@ -215,18 +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`. +/** Remove conjuncts that are unsafe to copy into another query tree (non-deterministic in this + * query, or stateful). 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`. */ -void removeExpressionsThatAreNotDeterministicInScopeOfQuery( - QueryTreeNodePtr & expression, - const ContextPtr & context); - -/** Remove conjuncts that call stateful functions (`aiEmbed`, `timeSeriesStoreTags`, and similar). - * Those can report `isDeterministicInScopeOfQuery` while still having side effects; JOIN filter - * pushdown refuses them via `ActionsDAG::hasStatefulFunctions`. - */ -void removeExpressionsThatAreStateful( +void removeExpressionsThatAreUnsafeToDuplicate( QueryTreeNodePtr & expression, const ContextPtr & context); diff --git a/src/Core/Joins.h b/src/Core/Joins.h index 1c863fcbb793..6973b1fd88df 100644 --- a/src/Core/Joins.h +++ b/src/Core/Joins.h @@ -140,6 +140,24 @@ enum class JoinTableSide : uint8_t const char * toString(JoinTableSide join_table_side); +/** Whether a post-JOIN `WHERE` conjunct on this side can be applied before the JOIN. + * Same rules as JOIN filter pushdown: 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`. + * Dictionary / lookup fill is a separate check (`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, diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 6fb7e7f88bc6..4f65042c5e6d 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -216,10 +216,7 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const } } -/// Same restrictions as JOIN filter pushdown (`filterPushDown.cpp`): -/// do not prefilter the null-producing side of an outer JOIN, the right side of -/// an `ASOF JOIN` (nearest-match would change), or either side of a `PASTE JOIN` -/// (positional alignment). +/// Same restrictions as JOIN filter pushdown (`canPrefilterJoinSide`). bool joinTreePreservesRowsForTable(const QueryTreeNodePtr & join_tree, const QueryTreeNodePtr & table) { std::vector stack = {join_tree}; @@ -235,13 +232,9 @@ bool joinTreePreservesRowsForTable(const QueryTreeNodePtr & join_tree, const Que const bool table_on_left = extractTableExpressionsSet(join->getLeftTableExpression()).contains(table.get()); const bool table_on_right = extractTableExpressionsSet(join->getRightTableExpression()).contains(table.get()); - if (isPaste(join->getKind()) && (table_on_left || table_on_right)) + if (table_on_left && !canPrefilterJoinSide(join->getKind(), join->getStrictness(), JoinTableSide::Left)) return false; - if (join->getStrictness() == JoinStrictness::Asof && table_on_right) - return false; - if (isRightOrFull(join->getKind()) && table_on_left) - return false; - if (isLeftOrFull(join->getKind()) && table_on_right) + if (table_on_right && !canPrefilterJoinSide(join->getKind(), join->getStrictness(), JoinTableSide::Right)) return false; stack.push_back(join->getLeftTableExpression()); stack.push_back(join->getRightTableExpression()); @@ -1040,8 +1033,7 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres { auto cloned = predicate->clone(); removeExpressionsThatDoNotDependOnTableIdentifiers(cloned, original_table_expression, query_context); - removeExpressionsThatAreNotDeterministicInScopeOfQuery(cloned, query_context); - removeExpressionsThatAreStateful(cloned, query_context); + removeExpressionsThatAreUnsafeToDuplicate(cloned, query_context); return cloned; }; diff --git a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp index 3e86fb520669..087b6c0a64ce 100644 --- a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp +++ b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -495,15 +496,24 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan:: const auto & left_stream_input_header = child->getInputHeaders().front(); const auto & right_stream_input_header = child->getInputHeaders().back(); - if (table_join_ptr && table_join_ptr->kind() == JoinKind::Full) - return 0; - if (logical_join && logical_join->getJoinOperator().kind == JoinKind::Full) - return 0; + JoinKind kind = JoinKind::Inner; + JoinStrictness strictness = JoinStrictness::Unspecified; + const bool have_join_kind = table_join_ptr || logical_join; + if (table_join_ptr) + { + kind = table_join_ptr->kind(); + strictness = table_join_ptr->strictness(); + } + else if (logical_join) + { + kind = logical_join->getJoinOperator().kind; + strictness = logical_join->getJoinOperator().strictness; + } - /// PASTE JOIN aligns rows from both sides by position, and pushing filters - /// to either side may change relative alignment - if ((table_join_ptr && table_join_ptr->kind() == JoinKind::Paste) - || (logical_join && logical_join->getJoinOperator().kind == JoinKind::Paste)) + /// `FULL` / `PASTE` cannot prefilter either side (`canPrefilterJoinSide`). + if (have_join_kind + && !canPrefilterJoinSide(kind, strictness, JoinTableSide::Left) + && !canPrefilterJoinSide(kind, strictness, JoinTableSide::Right)) return 0; std::unordered_map equivalent_left_stream_column_to_right_stream_column; @@ -646,29 +656,25 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan:: bool left_stream_filter_push_down_input_columns_available = true; bool right_stream_filter_push_down_input_columns_available = true; - if (table_join_ptr && table_join_ptr->kind() == JoinKind::Left) - right_stream_filter_push_down_input_columns_available = false; - else if (table_join_ptr && table_join_ptr->kind() == JoinKind::Right) - left_stream_filter_push_down_input_columns_available = false; - - if (logical_join && logical_join->getJoinOperator().kind == JoinKind::Left) - right_stream_filter_push_down_input_columns_available = false; - else if (logical_join && logical_join->getJoinOperator().kind == JoinKind::Right) - left_stream_filter_push_down_input_columns_available = false; + if (have_join_kind) + { + left_stream_filter_push_down_input_columns_available = canPrefilterJoinSide(kind, strictness, JoinTableSide::Left); + right_stream_filter_push_down_input_columns_available = canPrefilterJoinSide(kind, strictness, JoinTableSide::Right); + } - /** We disable push down to right table in cases: - * 1. Right side is already filled. Example: JOIN with Dictionary. - * 2. ASOF Right join is not supported. + /** We disable push down to the right table when the right side is already filled. + * Example: JOIN with Dictionary. `ASOF` is handled by `canPrefilterJoinSide`. */ - bool allow_push_down_to_right = join && join->allowPushDownToRight() && table_join_ptr && table_join_ptr->strictness() != JoinStrictness::Asof; if (logical_join) { bool has_logical_lookup = typeid_cast(child_node->children.back()->step.get()); - allow_push_down_to_right = !has_logical_lookup && logical_join->getJoinOperator().strictness != JoinStrictness::Asof; + if (has_logical_lookup) + right_stream_filter_push_down_input_columns_available = false; } - - if (!allow_push_down_to_right) + else if (!(join && join->allowPushDownToRight())) + { right_stream_filter_push_down_input_columns_available = false; + } Names equivalent_columns_to_push_down; @@ -891,7 +897,7 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan:: JoinKind::Left); } - if (join_filter_push_down_actions.right_stream_filter_to_push_down && allow_push_down_to_right) + if (join_filter_push_down_actions.right_stream_filter_to_push_down && right_stream_filter_push_down_input_columns_available) { if (logical_join) { From 20181690d76b34450f5fe4586d618008c037e289 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 16:19:53 +0200 Subject: [PATCH 10/14] Pin JOIN filter pushdown EXPLAIN test against parallel replicas and random settings `EXPLAIN` `Prewhere` is absent when parallel replicas rewrite the plan or when `optimize_move_to_prewhere` / `query_plan_optimize_prewhere` are randomized off. Co-authored-by: Cursor --- .../04673_join_filter_pushdown_count_subquery.sql | 6 ++++++ 1 file changed, 6 insertions(+) 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 index f0e7ebab062c..8699c29c74df 100644 --- a/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql +++ b/tests/queries/0_stateless/04673_join_filter_pushdown_count_subquery.sql @@ -1,3 +1,7 @@ +-- Tags: no-parallel-replicas, no-random-settings +-- no-parallel-replicas: EXPLAIN Prewhere differs with parallel replicas. +-- no-random-settings: `optimize_move_to_prewhere` / `query_plan_optimize_prewhere` are randomized off. + -- 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. @@ -30,6 +34,8 @@ SET enable_analyzer = 1; SET query_plan_filter_push_down = 1; SET enable_join_runtime_filters = 0; SET join_use_nulls = 1; +SET optimize_move_to_prewhere = 1; +SET query_plan_optimize_prewhere = 1; SELECT count() FROM From a1d45021599eb0c70c301732e78a23c3e7bad640 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 17:15:49 +0200 Subject: [PATCH 11/14] Drop non-function wrap predicates that depend on the other JOIN side `filterConjunctions` skipped a ColumnNode root, so `WHERE r.flag` was copied onto the `IStorageCluster` wrap and planning failed. Related: https://github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Analyzer/Utils.cpp | 4 ++ src/Analyzer/Utils.h | 3 +- ...uster_wrap_bare_other_side_where.reference | 1 + ...676_cluster_wrap_bare_other_side_where.sql | 37 +++++++++++++++++++ 4 files changed, 44 insertions(+), 1 deletion(-) create mode 100644 tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.reference create mode 100644 tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.sql diff --git a/src/Analyzer/Utils.cpp b/src/Analyzer/Utils.cpp index b539ae863ace..0c10ff31e6c8 100644 --- a/src/Analyzer/Utils.cpp +++ b/src/Analyzer/Utils.cpp @@ -1225,7 +1225,11 @@ void filterConjunctions( { auto * function = expression->as(); if (!function) + { + if (!keep(expression)) + expression = {}; return; + } if (function->getFunctionName() != "and") { diff --git a/src/Analyzer/Utils.h b/src/Analyzer/Utils.h index b6ba1cd8f8f6..93b24e2f0063 100644 --- a/src/Analyzer/Utils.h +++ b/src/Analyzer/Utils.h @@ -208,7 +208,8 @@ 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, diff --git a/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.reference b/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.reference new file mode 100644 index 000000000000..d00491fd7e5b --- /dev/null +++ b/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.reference @@ -0,0 +1 @@ +1 diff --git a/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.sql b/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.sql new file mode 100644 index 000000000000..17a77adf2ca2 --- /dev/null +++ b/tests/queries/0_stateless/04676_cluster_wrap_bare_other_side_where.sql @@ -0,0 +1,37 @@ +-- Tags: no-fasttest +-- no-fasttest: `fileCluster` is not in the fast test build. +-- A wrap `WHERE` that is a bare column from the other JOIN side must be dropped +-- rather than copied onto `SELECT cols FROM fileCluster`. Otherwise planning +-- fails with an unknown identifier. + +SET enable_analyzer = 1; +SET query_plan_filter_push_down = 1; +SET query_plan_join_swap_table = 0; +SET enable_join_runtime_filters = 0; +SET enable_parallel_replicas = 0; + +INSERT INTO FUNCTION file(currentDatabase() || '_04676_wrap_left.tsv', 'TSV', 'n Int32') +SELECT number + 1 +FROM numbers(2) +SETTINGS engine_file_truncate_on_insert = 1; + +DROP TABLE IF EXISTS t_04676_right; +CREATE TABLE t_04676_right +( + id Int32, + flag UInt8 +) +ENGINE = Memory; +INSERT INTO t_04676_right VALUES (1, 1), (2, 0); + +SELECT l.n +FROM fileCluster( + 'test_cluster_one_shard_two_replicas', + currentDatabase() || '_04676_wrap_left.tsv', + 'TSV', + 'n Int32') AS l +INNER JOIN t_04676_right AS r ON l.n = r.id +WHERE r.flag +ORDER BY l.n; + +DROP TABLE t_04676_right; From cdc4b28a6ddbf7c8cad0ae93ff2bc9b190937083 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 19:29:52 +0200 Subject: [PATCH 12/14] Keep equivalent-key JOIN filter pushdown separate from side prefilter `canPrefilterJoinSide` only marks ordinary columns unavailable on the null-producing side. Attaching a rewritten equivalent-key filter to that child is still `allow_push_down_to_right` (dictionary / lookup / `ASOF`). Related: https://github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Core/Joins.h | 9 ++++---- .../Optimizations/filterPushDown.cpp | 21 ++++++++++++------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/src/Core/Joins.h b/src/Core/Joins.h index 6973b1fd88df..83373bb0fb1f 100644 --- a/src/Core/Joins.h +++ b/src/Core/Joins.h @@ -140,10 +140,11 @@ enum class JoinTableSide : uint8_t const char * toString(JoinTableSide join_table_side); -/** Whether a post-JOIN `WHERE` conjunct on this side can be applied before the JOIN. - * Same rules as JOIN filter pushdown: 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`. - * Dictionary / lookup fill is a separate check (`JoinStep::allowPushDownToRight`). +/** 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) { diff --git a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp index 087b6c0a64ce..a8ce015a2d7e 100644 --- a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp +++ b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp @@ -662,19 +662,24 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan:: right_stream_filter_push_down_input_columns_available = canPrefilterJoinSide(kind, strictness, JoinTableSide::Right); } - /** We disable push down to the right table when the right side is already filled. - * Example: JOIN with Dictionary. `ASOF` is handled by `canPrefilterJoinSide`. + /** `canPrefilterJoinSide` only decides whether this side's own columns may be + * used as ordinary filter inputs (false on the null-producing outer-JOIN side). + * Equivalent-key filters can still be attached to that child. That attach is + * gated by `allow_push_down_to_right`: + * 1. Right side is already filled. Example: JOIN with Dictionary. + * 2. `ASOF` right join is not supported. */ + bool allow_push_down_to_right = join && join->allowPushDownToRight() && table_join_ptr + && table_join_ptr->strictness() != JoinStrictness::Asof; if (logical_join) { bool has_logical_lookup = typeid_cast(child_node->children.back()->step.get()); - if (has_logical_lookup) - right_stream_filter_push_down_input_columns_available = false; + allow_push_down_to_right = !has_logical_lookup + && logical_join->getJoinOperator().strictness != JoinStrictness::Asof; } - else if (!(join && join->allowPushDownToRight())) - { + + if (!allow_push_down_to_right) right_stream_filter_push_down_input_columns_available = false; - } Names equivalent_columns_to_push_down; @@ -897,7 +902,7 @@ static size_t tryPushDownOverJoinStep(QueryPlan::Node * parent_node, QueryPlan:: JoinKind::Left); } - if (join_filter_push_down_actions.right_stream_filter_to_push_down && right_stream_filter_push_down_input_columns_available) + if (join_filter_push_down_actions.right_stream_filter_to_push_down && allow_push_down_to_right) { if (logical_join) { From d9ebb0b72e92c312f2e770ca0f5636a8a4c0fd1a Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 24 Aug 2026 21:48:45 +0200 Subject: [PATCH 13/14] Rebuild `IStorageCluster` listing when a later `applyFilters` predicate changes The cluster JOIN wrap applies the copied `WHERE` before the optimizer can push an outer filter, so listing used to keep `a` and miss `a AND b`. https: //github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 31 +++++++++++++++---- src/Storages/IStorageCluster.h | 2 ++ ...test_cluster_join_filter_minmax_pruning.py | 25 ++++++++++++++- 3 files changed, 51 insertions(+), 7 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 4fec16100b6d..e134491bd6df 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -98,18 +98,36 @@ void ReadFromCluster::applyFilters(ActionDAGNodes added_filter_nodes) void ReadFromCluster::createExtension(const ActionsDAG::Node * predicate) { - /// 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; + const ActionsDAG * filter = filter_actions_dag ? filter_actions_dag.get() : query_info.filter_actions_dag.get(); + const UInt64 filter_hash = filter ? filter->getHash() : 0; + const bool has_predicate = predicate != nullptr; + + if (extension) + { + /// Remote sources already hold the iterator; replacing it would be a use-after-free. + if (extension_used_in_pipeline) + return; + + /// Optimizer `applyFilters` with no extra FilterSteps must not replace a + /// listing that already has a predicate with an empty one. + if (!has_predicate) + return; + + /// Same listing predicate. Recreate when a later `applyFilters` replaces + /// `a` with `a AND b` (cluster JOIN wrap applies the copied `WHERE` + /// before the optimizer pushes outer filters). + if (extension_has_predicate && filter_hash == extension_filter_hash) + return; + } extension = storage->getTaskIteratorExtension( predicate, - filter_actions_dag ? filter_actions_dag.get() : query_info.filter_actions_dag.get(), + filter, context, cluster, getStorageSnapshot()->metadata); - extension_has_predicate = predicate != nullptr; + extension_has_predicate = has_predicate; + extension_filter_hash = filter_hash; } namespace @@ -602,6 +620,7 @@ void ReadFromCluster::initializePipeline(QueryPipelineBuilder & pipeline, const 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); + extension_used_in_pipeline = true; ProfileEvents::increment(ProfileEvents::Shards, max_replicas_to_use); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index e9714bf7f694..6d725f1c2f94 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -173,6 +173,8 @@ class ReadFromCluster : public SourceStepWithFilter std::optional extension; bool extension_has_predicate = false; + UInt64 extension_filter_hash = 0; + bool extension_used_in_pipeline = 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 index fb1e5009d89a..ce3f0411485a 100644 --- 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 @@ -44,6 +44,9 @@ def execute_spark_query(query: str): 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)") + # Passes `bid >= 3`, fails `datetime >= 2024-01-03`. Distinguishes listing + # that only saw the inner JOIN `WHERE` from listing that also got the outer `WHERE`. + execute_spark_query(f"INSERT INTO {TABLE_NAME} VALUES (DATE '2024-01-01', 'AAPL', 4)") iceberg = get_creation_expression( storage_type, @@ -83,7 +86,7 @@ def check_validity_and_get_prunned_files(select_expression): select_expression, ) - # Three data files with disjoint bid ranges; bid >= 3 keeps one file. + # Four data files: bid 1/2/3/4. `bid >= 3` keeps two files (prunes 2). expected_pruned = 2 assert ( @@ -127,3 +130,23 @@ def check_validity_and_get_prunned_files(select_expression): ) == expected_pruned ) + + # Inner `bid >= 3` is copied onto the cluster wrap during planning. The outer + # `datetime` predicate is pushed later; listing must be rebuilt or the extra + # file with bid=4 / datetime=2024-01-01 is not 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 + ) + WHERE datetime >= '2024-01-03' + """ + ) + == 3 + ) From cf210cbe24b2340e854779d2626e2e99ed8eebf7 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 25 Aug 2026 01:16:38 +0200 Subject: [PATCH 14/14] Do not copy server-constant wrap predicates such as `hostName` The wrap `WHERE` is sent to remote cluster nodes, where `hostName` can differ from the initiator and drop every row. https: //github.com/Altinity/ClickHouse/pull/2249 Co-authored-by: Cursor --- src/Analyzer/Utils.cpp | 4 ++- src/Analyzer/Utils.h | 4 ++- ...ster_wrap_server_constant_filter.reference | 1 + ...77_cluster_wrap_server_constant_filter.sql | 35 +++++++++++++++++++ 4 files changed, 42 insertions(+), 2 deletions(-) create mode 100644 tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.reference create mode 100644 tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.sql diff --git a/src/Analyzer/Utils.cpp b/src/Analyzer/Utils.cpp index 0c10ff31e6c8..fab99e9cd1ab 100644 --- a/src/Analyzer/Utils.cpp +++ b/src/Analyzer/Utils.cpp @@ -1214,7 +1214,9 @@ bool isSafeToDuplicateInQueryTree(const QueryTreeNodePtr & node) node, [](const FunctionBasePtr & function_base) { - return function_base->isDeterministicInScopeOfQuery() && !function_base->isStateful(); + return function_base->isDeterministicInScopeOfQuery() + && !function_base->isStateful() + && !function_base->isServerConstant(); }); } diff --git a/src/Analyzer/Utils.h b/src/Analyzer/Utils.h index 93b24e2f0063..86f06fe688fb 100644 --- a/src/Analyzer/Utils.h +++ b/src/Analyzer/Utils.h @@ -217,9 +217,11 @@ void removeExpressionsThatDoNotDependOnTableIdentifiers( const ContextPtr & context); /** Remove conjuncts that are unsafe to copy into another query tree (non-deterministic in this - * query, or stateful). Nested `and` is flattened the same way as + * 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`. + * Server constants such as `hostName` must stay on the initiator: the wrap `WHERE` is sent to + * remote cluster nodes, where those functions can return a different value. */ void removeExpressionsThatAreUnsafeToDuplicate( QueryTreeNodePtr & expression, diff --git a/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.reference b/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.reference new file mode 100644 index 000000000000..0cfbf08886fc --- /dev/null +++ b/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.reference @@ -0,0 +1 @@ +2 diff --git a/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.sql b/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.sql new file mode 100644 index 000000000000..34cb451d72ef --- /dev/null +++ b/tests/queries/0_stateless/04677_cluster_wrap_server_constant_filter.sql @@ -0,0 +1,35 @@ +-- Tags: no-fasttest +-- no-fasttest: `fileCluster` is not in the fast test build. +-- `hostName` is `isServerConstant`. Copying it into the `IStorageCluster` wrap +-- would evaluate it on remotes, where it can differ from the initiator, and +-- drop every row. The `n < 2` conjunct must still be copied. + +SET enable_analyzer = 1; +SET query_plan_filter_push_down = 1; +SET query_plan_join_swap_table = 0; +SET enable_join_runtime_filters = 0; +SET enable_parallel_replicas = 0; + +INSERT INTO FUNCTION file(currentDatabase() || '_04677_wrap_left.tsv', 'TSV', 'n UInt64') +SELECT number +FROM numbers(3) +SETTINGS engine_file_truncate_on_insert = 1; + +DROP TABLE IF EXISTS t_04677_right; +CREATE TABLE t_04677_right +( + n UInt64 +) +ENGINE = Memory; +INSERT INTO t_04677_right VALUES (0), (1), (2); + +SELECT count() +FROM fileCluster( + 'test_cluster_one_shard_two_replicas', + currentDatabase() || '_04677_wrap_left.tsv', + 'TSV', + 'n UInt64') AS l +LEFT JOIN t_04677_right AS r ON l.n = r.n +WHERE l.n < 2 AND hostName() = hostName(); + +DROP TABLE t_04677_right;