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..12e25b605fb4 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,49 @@ void checkAccessRightsForSubquery(const QueryTreeNodePtr & subquery_node, const } } +/// `IStorageCluster` JOINs wrap the left table in a subquery. Attach dummy-analysis +/// filters to the wrap source for listing only; do not add a FilterStep, which would +/// drop unused columns from the wrap header. +void tryAddClusterWrapFilter(QueryPlan & query_plan, const TableExpressionData & table_expression_data) +{ + const auto & filter_actions = table_expression_data.getFilterActions(); + if (!filter_actions || !query_plan.isInitialized()) + return; + + QueryPlan::Node * node = query_plan.getRootNode(); + while (node && !node->children.empty()) + node = node->children.front(); + + auto * source = node ? dynamic_cast(node->step.get()) : nullptr; + if (!source) + return; + + auto filter_dag = filter_actions->clone(); + const auto filter_column_name = filter_dag.getOutputs().at(0)->result_name; + const auto & header = source->getOutputHeader(); + ActionsDAG rename_dag(header->getColumnsWithTypeAndName()); + const auto & identifier_to_name = table_expression_data.getColumnIdentifierToColumnName(); + + for (const auto * input : filter_dag.getInputs()) + { + if (header->has(input->result_name)) + continue; + + auto it = identifier_to_name.find(input->result_name); + if (it == identifier_to_name.end() || !header->has(it->second)) + continue; + + const auto & physical = rename_dag.findInOutputs(it->second); + rename_dag.addOrReplaceInOutputs(rename_dag.addAlias(physical, input->result_name)); + } + + filter_dag = ActionsDAG::merge(std::move(rename_dag), std::move(filter_dag)); + source->addFilter(std::move(filter_dag), filter_column_name); + /// Wrap subquery planning already called `applyFilters` with no predicate. + /// Apply now so icebergCluster listing is recreated with the WHERE. + source->SourceStepWithFilterBase::applyFilters(); +} + bool shouldIgnoreQuotaAndLimits(const TableNode & table_node) { const auto & storage_id = table_node.getStorageID(); @@ -920,8 +964,46 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres if (wrap_read_columns_in_subquery) { + auto original_table_expression = table_expression; + + /// Subqueries inherit the outer GlobalPlannerContext, whose filter map is keyed by + /// outer table nodes. Collect filters for this JOIN query so icebergCluster listing + /// still sees left-only WHERE after the wrap. + if (!table_expression_data.getFilterActions() && select_query_info.query_tree) + { + auto collected = collectFiltersForAnalysis(select_query_info.query_tree, select_query_options, nullptr); + auto it = collected.find(table_expression); + if (it != collected.end() && it->second.filter_actions) + table_expression_data.setFilterActions(it->second.filter_actions->clone()); + } + auto columns = table_expression_data.getColumns(); - table_expression = buildSubqueryToReadColumnsFromTableExpression(columns, table_expression, query_context); + table_expression = buildSubqueryToReadColumnsFromTableExpression(columns, original_table_expression, query_context); + + /// Wrap is planned as `SELECT cols FROM icebergCluster` with no JOIN. Copy left-only + /// WHERE/PREWHERE so initiator file listing sees the same predicate as a single-table + /// `icebergCluster` read. Same helper as `IStorageCluster::updateQueryWithJoinToSendIfNeeded`. + if (const auto * parent_query = select_query_info.query_tree->as()) + { + 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); + } + } } auto * table_node = table_expression->as(); @@ -1491,12 +1573,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 +1589,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/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/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 + ) 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;