From 01e19fcc62a49542c842b3c3d845c746eac3422e Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 26 Oct 2026 16:59:52 +0100 Subject: [PATCH 1/9] Add `05428` for Iceberg partition `min`, `max`, and grouped `count` from manifests. The shortcut has to answer those aggregates from manifest `record_count` for `IcebergLocal`, `IcebergS3`, `Iceberg`, their table functions, and `icebergS3Cluster`, including a partition range and a subquery. The test fails until that read exists. Co-authored-by: Cursor --- ...erg_partition_aggregate_metadata.reference | 385 ++++++++++++++++++ ...28_iceberg_partition_aggregate_metadata.sh | 195 +++++++++ 2 files changed, 580 insertions(+) create mode 100644 tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference create mode 100755 tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference new file mode 100644 index 0000000000000..6ecff4cb2cc0b --- /dev/null +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -0,0 +1,385 @@ +IcebergLocal +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +icebergLocal +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +IcebergS3 +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +icebergS3 +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +icebergS3Cluster +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +Iceberg +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data +iceberg +files 10 +max_d +2026-01-10 +min_d +2026-01-01 +max_d_filter +2026-01-10 +max_d_range +2026-01-10 +groupby_d +2026-01-01 1000 +2026-01-02 1000 +2026-01-03 1000 +2026-01-04 1000 +2026-01-05 1000 +2026-01-06 1000 +2026-01-07 1000 +2026-01-08 1000 +2026-01-09 1000 +2026-01-10 1000 +groupby_d_range +2026-01-09 1000 +2026-01-10 1000 +max_of_groupby_count +2026-01-10 +max_of_groupby +2026-01-10 +groupby_d_only +2026-01-01 +2026-01-02 +2026-01-03 +2026-01-04 +2026-01-05 +2026-01-06 +2026-01-07 +2026-01-08 +2026-01-09 +2026-01-10 +count +10000 +max_k +99 +reads +max_d metadata +min_d metadata +max_d_filter metadata +max_d_range metadata +groupby_d metadata +groupby_d_range metadata +max_of_groupby_count metadata +max_of_groupby metadata +groupby_d_only metadata +count metadata +max_k data diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh new file mode 100755 index 0000000000000..321ece3c29a3a --- /dev/null +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -0,0 +1,195 @@ +#!/usr/bin/env bash +# Tags: no-fasttest, no-replicated-database +# - no-fasttest: Iceberg needs Avro, and the S3 engines need MinIO. +# - no-replicated-database: these engines are not replicated. +# +# The partition value and `record_count` live in the Iceberg manifest, so the same shortcut +# applies to every engine that reads through `IcebergMetadata`: the local table, the S3 table, +# the generic `Iceberg` table, their table functions, and `icebergS3Cluster`. A plain `S3` +# table has no manifest and is not covered here. +# +# On a cluster the initiator's `query_log.read_rows` is the sum across shards. The answer has +# to come from one metadata pass: two shards each returning the whole table would double +# `count` and the grouped counts. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +set -euo pipefail + +PREFIX="${CLICKHOUSE_TEST_UNIQUE_NAME}" +LOCAL_TABLE="t_${PREFIX}_local" +S3_TABLE="t_${PREFIX}_s3" +ICEBERG_TABLE="t_${PREFIX}_ice" +LOCAL_PATH="${USER_FILES_PATH}/${LOCAL_TABLE}/" +S3_KEY="${PREFIX}/s3" +ICEBERG_KEY="${PREFIX}/ice" +ICEBERG_URL="http://localhost:11111/test/${ICEBERG_KEY}/" + +cleanup() +{ + ${CLICKHOUSE_CLIENT} --query " + DROP TABLE IF EXISTS ${LOCAL_TABLE}; + DROP TABLE IF EXISTS ${S3_TABLE}; + DROP TABLE IF EXISTS ${ICEBERG_TABLE}" || true + rm -rf "${LOCAL_PATH}" +} +trap cleanup EXIT + +data_files() +{ + ${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT count() + FROM system.iceberg_files + WHERE database = currentDatabase() AND table = '${1}' AND toString(content) = 'DATA'" +} + +create_table() +{ + local table=$1 + local engine=$2 + ${CLICKHOUSE_CLIENT} --query " + CREATE TABLE ${table} (d Date32, k Int64, v Float64) + ENGINE = ${engine} + PARTITION BY d + SETTINGS allow_experimental_iceberg_compaction = 0" + ${CLICKHOUSE_CLIENT} --query " + INSERT INTO ${table} + SELECT toDate32('2026-01-01') + intDiv(number, 1000), number % 100, 0 + FROM numbers(10000) + SETTINGS allow_insert_into_iceberg = 1, async_insert = 0" + local files + files=$(data_files "${table}") + if [ "${files}" -ne 10 ]; then + echo "${table}: expected 10 data files, got ${files}" + exit 1 + fi +} + +run() +{ + local tag=$1 + local sql=$2 + ${CLICKHOUSE_CLIENT} --format TabSeparated --query "${sql} +SETTINGS + log_comment = '${tag}', + log_queries = 1, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + use_iceberg_partition_pruning = 1, + enable_parallel_replicas = 0" +} + +# One metadata pass reads at most the data files selected by the predicate. +# `max(k)` is not a partition column, so it still reads the 10000 data rows. +exercise() +{ + local label=$1 + local source=$2 + local tag="${PREFIX}_${label}" + + echo "${label}" + echo "files 10" + + echo "max_d" + run "${tag}_max_d" "SELECT max(d) FROM ${source}" + + echo "min_d" + run "${tag}_min_d" "SELECT min(d) FROM ${source}" + + # The predicate matches every partition. It must not force the data scan back on. + echo "max_d_filter" + run "${tag}_max_d_filter" "SELECT max(d) FROM ${source} WHERE d >= '2026-01-01'" + + # Two dates, 2026-01-09 and 2026-01-10. Two files, not the 2000 rows inside them. + echo "max_d_range" + run "${tag}_max_d_range" "SELECT max(d) FROM ${source} WHERE d >= '2026-01-09'" + + echo "groupby_d" + run "${tag}_groupby_d" "SELECT concat(toString(d), ' ', toString(count())) FROM ${source} GROUP BY d ORDER BY d" + + echo "groupby_d_range" + run "${tag}_groupby_d_range" "SELECT concat(toString(d), ' ', toString(count())) FROM ${source} WHERE d >= '2026-01-09' GROUP BY d ORDER BY d" + + echo "max_of_groupby_count" + run "${tag}_max_of_groupby_count" "SELECT max(d) FROM (SELECT d, count(*) FROM ${source} GROUP BY d)" + + echo "max_of_groupby" + run "${tag}_max_of_groupby" "SELECT max(d) FROM (SELECT d FROM ${source} GROUP BY d)" + + echo "groupby_d_only" + run "${tag}_groupby_d_only" "SELECT d FROM ${source} GROUP BY d ORDER BY d" + + echo "count" + run "${tag}_count" "SELECT count() FROM ${source}" + + echo "max_k" + run "${tag}_max_k" "SELECT max(k) FROM ${source}" + + ${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" + + echo "reads" + ${CLICKHOUSE_CLIENT} --format TabSeparated --query " + WITH entries AS + ( + SELECT + replaceOne(log_comment, '${tag}_', '') AS name, + max(read_rows) AS read_rows, + max(ProfileEvents['IcebergTrivialCountOptimizationApplied']) AS trivial_count + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND startsWith(log_comment, '${tag}_') + GROUP BY name + ), + expected AS + ( + SELECT 'max_d' AS name, 10 AS lim, 0 AS ord + UNION ALL SELECT 'min_d', 10, 1 + UNION ALL SELECT 'max_d_filter', 10, 2 + UNION ALL SELECT 'max_d_range', 2, 3 + UNION ALL SELECT 'groupby_d', 10, 4 + UNION ALL SELECT 'groupby_d_range', 2, 5 + UNION ALL SELECT 'max_of_groupby_count', 10, 6 + UNION ALL SELECT 'max_of_groupby', 10, 7 + UNION ALL SELECT 'groupby_d_only', 10, 8 + UNION ALL SELECT 'count', 10, 9 + UNION ALL SELECT 'max_k', 0, 10 + ) + SELECT concat(e.name, ' ', multiIf( + e.name = 'max_k', + if(q.read_rows >= 10000, 'data', concat('metadata read_rows=', toString(q.read_rows))), + e.name = 'count', + if( + q.read_rows <= e.lim AND q.trivial_count = 1, + 'metadata', + concat( + 'data read_rows=', toString(q.read_rows), + ' limit=', toString(e.lim), + ' trivial_count=', toString(q.trivial_count))), + if( + q.read_rows <= e.lim, + 'metadata', + concat( + 'data read_rows=', toString(q.read_rows), + ' limit=', toString(e.lim))))) + FROM expected AS e + INNER JOIN entries AS q ON e.name = q.name + ORDER BY e.ord + SETTINGS enable_parallel_replicas = 0" +} + +create_table "${LOCAL_TABLE}" "IcebergLocal('${LOCAL_PATH}')" +exercise "IcebergLocal" "${LOCAL_TABLE}" +exercise "icebergLocal" "icebergLocal('${LOCAL_PATH}')" + +create_table "${S3_TABLE}" "IcebergS3(s3_conn, filename = '${S3_KEY}')" +exercise "IcebergS3" "${S3_TABLE}" +exercise "icebergS3" "icebergS3(s3_conn, filename = '${S3_KEY}')" +exercise "icebergS3Cluster" "icebergS3Cluster('test_cluster_two_shards_localhost', s3_conn, filename = '${S3_KEY}')" + +create_table "${ICEBERG_TABLE}" "Iceberg('${ICEBERG_URL}', 'test', 'testtest')" +exercise "Iceberg" "${ICEBERG_TABLE}" +exercise "iceberg" "iceberg('${ICEBERG_URL}', 'test', 'testtest', storage_type = 's3')" From d381d8c6292ab30cc8d25f2f32517ea0db9fdeaa Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 26 Oct 2026 18:20:55 +0100 Subject: [PATCH 2/9] Answer Iceberg identity partition `min`, `max`, and grouped `count` from manifests The partition value and `record_count` are already in the manifest, so these aggregates do not scan the data files. Plain `count()` stays on the trivial-count path. The generic `Iceberg` engine forwards `isIcebergConfiguration` so `system.iceberg_files` and this shortcut see the real backend. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 492 +++++++++++++++++- src/Storages/IStorage.h | 14 + .../DataLakes/DataLakeConfiguration.h | 3 + .../DataLakes/Iceberg/IcebergMetadata.cpp | 72 +++ .../DataLakes/Iceberg/IcebergMetadata.h | 5 + .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 6 +- .../ObjectStorage/DataLakes/Iceberg/Utils.h | 4 +- .../ObjectStorage/StorageObjectStorage.cpp | 24 + .../ObjectStorage/StorageObjectStorage.h | 1 + .../StorageObjectStorageCluster.cpp | 9 + .../StorageObjectStorageCluster.h | 1 + 11 files changed, 626 insertions(+), 5 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index e5cf42e3d566e..44eb9cab3f4a0 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -21,6 +21,19 @@ #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + #include #include @@ -709,6 +722,464 @@ RowPolicyFilterPtr getEffectiveRowPolicyFilter(const StoragePtr & storage, const return row_policy_filter; } +/// `min` / `max` / `count` of identity partition columns, with an optional filter on those columns only. +/// The manifest stores the partition value and `record_count` per data file, so the aggregate does not +/// read the files. The result is one mergeable aggregation state per group. Runs on the initiator only: +/// a secondary query must not answer from the whole snapshot, or a cluster would count every shard twice. +bool applyIcebergIdentityPartitionAggregationIfPossible( + QueryPlan & query_plan, + SelectQueryInfo & select_query_info, + const TableNode * table_node, + const TableFunctionNode * table_function_node, + const QueryTreeNodePtr & query_tree, + ContextMutablePtr & query_context, + const PlannerContextPtr & planner_context) +{ + if (query_plan.isInitialized()) + return false; + + const auto & settings = query_context->getSettingsRef(); + /// `ReadFromPreparedSource` cannot be cloned by the Cascades optimizer. + if (settings[Setting::make_distributed_plan] && settings[Setting::enable_cascades_optimizer]) + return false; + + /// With this setting an empty input must produce no row. The prepared state always produces one + /// row for an aggregate without `GROUP BY`, so leave that case to the normal pipeline. + if (settings[Setting::empty_result_for_aggregation_by_empty_set]) + return false; + + if (query_context->getClientInfo().query_kind == ClientInfo::QueryKind::SECONDARY_QUERY) + return false; + + if (settings[Setting::allow_experimental_parallel_reading_from_replicas] > 0 && settings[Setting::max_parallel_replicas] > 1) + { + if (settings[Setting::parallel_replicas_mode] == ParallelReplicasMode::CUSTOM_KEY_RANGE + || settings[Setting::parallel_replicas_mode] == ParallelReplicasMode::CUSTOM_KEY_SAMPLING + || settings[Setting::parallel_replicas_mode] == ParallelReplicasMode::SAMPLING_KEY) + return false; + } + + const auto & storage = table_node ? table_node->getStorage() : table_function_node->getStorage(); + if (!storage) + return false; + + if (getEffectiveRowPolicyFilter(storage, query_context)) + return false; + + if (select_query_info.additional_filter_ast) + return false; + + if (query_context->getCurrentTransaction()) + return false; + + if (hasTrivialCountIncompatibleModifiers(table_node, table_function_node)) + return false; + + auto & query_node = query_tree->as(); + if (query_node.hasPrewhere() || query_node.hasWindow() || query_node.hasQualify()) + return false; + + if (query_node.isGroupByWithTotals() || query_node.isGroupByWithRollup() || query_node.isGroupByWithCube() + || query_node.isGroupByWithGroupingSets()) + return false; + + if (!query_node.getCorrelatedColumnsSet().empty() || queryHasArrayJoinInJoinTree(query_tree)) + return false; + + struct PlainColumn + { + String schema_name; + String action_name; + DataTypePtr type; + }; + + auto appendColumn = [&](const QueryTreeNodePtr & node, std::vector & out) -> bool + { + const auto * column = node->as(); + if (!column || column->hasExpression()) + return false; + String action_name = calculateActionNodeName(node, *planner_context); + if (action_name.empty()) + return false; + out.push_back(PlainColumn{column->getColumnName(), std::move(action_name), column->getColumnType()}); + return true; + }; + + auto collectColumns = [&](const QueryTreeNodePtr & node, std::vector & out, auto & self) -> bool + { + if (!node) + return true; + + switch (node->getNodeType()) + { + case QueryTreeNodeType::COLUMN: + return appendColumn(node, out); + case QueryTreeNodeType::CONSTANT: + return true; + case QueryTreeNodeType::FUNCTION: + { + const auto & function = node->as(); + if (!function.isOrdinaryFunction() || function.getFunctionName() == "arrayJoin") + return false; + for (const auto & argument : function.getArguments().getNodes()) + { + if (!self(argument, out, self)) + return false; + } + return true; + } + case QueryTreeNodeType::LIST: + { + for (const auto & child : node->as().getNodes()) + { + if (!self(child, out, self)) + return false; + } + return true; + } + default: + return false; + } + }; + + enum class AggKind + { + CountAll, + CountColumn, + Min, + Max + }; + + struct AggSpec + { + AggKind kind; + String schema_name; + String action_name; + AggregateFunctionPtr function; + DataTypePtr state_type; + }; + + std::vector referenced_columns; + std::vector aggregates; + NameSet aggregate_names; + for (const auto & aggregate_node : collectAggregateFunctionNodes(query_tree)) + { + const auto & function = aggregate_node->as(); + if (!function.isAggregateFunction() || !function.getAggregateFunction()) + return false; + if (!function.getParameters().getNodes().empty() || !function.getAggregateFunction()->getParameters().empty()) + return false; + + AggSpec spec; + spec.function = function.getAggregateFunction(); + spec.action_name = calculateActionNodeName(aggregate_node, *planner_context); + if (spec.action_name.empty()) + return false; + if (!aggregate_names.emplace(spec.action_name).second) + continue; + + spec.state_type = std::make_shared( + spec.function, spec.function->getArgumentTypes(), spec.function->getParameters()); + + const auto & name = spec.function->getName(); + const auto & arguments = function.getArguments().getNodes(); + if (name == "count" && arguments.empty()) + { + spec.kind = AggKind::CountAll; + } + else if (name == "count" && arguments.size() == 1) + { + if (!appendColumn(arguments.front(), referenced_columns)) + return false; + spec.kind = AggKind::CountColumn; + spec.schema_name = referenced_columns.back().schema_name; + } + else if ((name == "min" || name == "max") && arguments.size() == 1) + { + if (!appendColumn(arguments.front(), referenced_columns)) + return false; + spec.kind = name == "min" ? AggKind::Min : AggKind::Max; + spec.schema_name = referenced_columns.back().schema_name; + } + else + { + return false; + } + aggregates.push_back(std::move(spec)); + } + + std::vector group_keys; + NameSet key_names; + if (query_node.hasGroupBy()) + { + for (const auto & key_node : query_node.getGroupBy().getNodes()) + { + if (!appendColumn(key_node, group_keys)) + return false; + if (!key_names.emplace(group_keys.back().action_name).second) + { + group_keys.pop_back(); + continue; + } + referenced_columns.push_back(group_keys.back()); + } + } + + if (aggregates.empty() && group_keys.empty()) + return false; + + std::vector where_columns; + if (query_node.hasWhere()) + { + if (!collectColumns(query_node.getWhere(), where_columns, collectColumns)) + return false; + referenced_columns.insert(referenced_columns.end(), where_columns.begin(), where_columns.end()); + } + + auto files = storage->getIdentityPartitionFiles(query_context); + if (!files) + return false; + + auto findIdentity = [](const IdentityPartitionFile & file, const String & name) -> const Field * + { + for (const auto & [column_name, value] : file.identity_values) + { + if (column_name == name) + return &value; + } + return nullptr; + }; + + for (const auto & column : referenced_columns) + { + for (const auto & file : *files) + { + const Field * value = findIdentity(file, column.schema_name); + if (!value) + return false; + if (value->isNull() && !column.type->isNullable()) + return false; + } + } + + std::vector kept; + kept.reserve(files->size()); + if (query_node.hasWhere() && !files->empty()) + { + std::vector filter_columns; + for (const auto & column : where_columns) + { + bool seen = false; + for (const auto & existing : filter_columns) + { + if (existing.action_name == column.action_name) + { + seen = true; + break; + } + } + if (!seen) + filter_columns.push_back(column); + } + + ColumnsWithTypeAndName input_columns; + Block filter_block; + for (const auto & column : filter_columns) + { + auto values = column.type->createColumn(); + for (const auto & file : *files) + { + const Field * value = findIdentity(file, column.schema_name); + if (!value) + return false; + values->insert(*value); + } + input_columns.emplace_back(nullptr, column.type, column.action_name); + filter_block.insert(ColumnWithTypeAndName(std::move(values), column.type, column.action_name)); + } + if (filter_block.columns() == 0) + { + auto dummy = ColumnUInt8::create(files->size()); + filter_block.insert(ColumnWithTypeAndName(std::move(dummy), std::make_shared(), "_dummy")); + } + + ColumnNodePtrWithHashSet no_correlated_columns; + auto [filter_dag, correlated] = buildActionsDAGFromExpressionNode( + query_node.getWhere(), input_columns, planner_context, no_correlated_columns); + if (correlated.notEmpty() || filter_dag.getOutputs().size() != 1) + return false; + if (!WhichDataType(removeNullable(filter_dag.getOutputs().front()->result_type)).isUInt8()) + return false; + + const String & filter_name = filter_dag.getOutputs().front()->result_name; + auto actions = std::make_shared(std::move(filter_dag)); + size_t num_rows = filter_block.rows(); + actions->execute(filter_block, num_rows); + if (!filter_block.has(filter_name)) + return false; + + const IColumn & filter_column = *filter_block.getByName(filter_name).column; + if (filter_column.size() != 1 && filter_column.size() != files->size()) + return false; + + auto keeps = [](const IColumn & column, size_t row, auto & self) -> bool + { + if (const auto * constant = checkAndGetColumn(&column)) + return self(constant->getDataColumn(), 0, self); + if (const auto * nullable = checkAndGetColumn(&column)) + { + if (nullable->isNullAt(row)) + return false; + return nullable->getNestedColumn().getUInt(row) != 0; + } + return column.getUInt(row) != 0; + }; + + for (size_t i = 0; i < files->size(); ++i) + { + const size_t row = filter_column.size() == 1 ? 0 : i; + if (keeps(filter_column, row, keeps)) + kept.push_back(i); + } + } + else + { + for (size_t i = 0; i < files->size(); ++i) + kept.push_back(i); + } + + struct Group + { + std::vector keys; + UInt64 rows = 0; + std::vector column_counts; + std::vector> extremal_values; + }; + + std::map, size_t> group_positions; + std::vector groups; + if (group_keys.empty()) + { + Group group; + group.column_counts.assign(aggregates.size(), 0); + group.extremal_values.resize(aggregates.size()); + groups.push_back(std::move(group)); + } + + for (size_t index : kept) + { + const auto & file = (*files)[index]; + std::vector key; + key.reserve(group_keys.size()); + for (const auto & group_key : group_keys) + { + const Field * value = findIdentity(file, group_key.schema_name); + if (!value) + return false; + key.push_back(*value); + } + + size_t group_index = 0; + if (!group_keys.empty()) + { + auto [it, inserted] = group_positions.emplace(key, groups.size()); + if (inserted) + { + Group group; + group.keys = it->first; + group.column_counts.assign(aggregates.size(), 0); + group.extremal_values.resize(aggregates.size()); + groups.push_back(std::move(group)); + } + group_index = it->second; + } + + Group & group = groups[group_index]; + if (common::addOverflow(group.rows, file.record_count, group.rows)) + return false; + + for (size_t agg_index = 0; agg_index < aggregates.size(); ++agg_index) + { + const auto & aggregate = aggregates[agg_index]; + if (aggregate.kind == AggKind::CountAll) + continue; + + const Field * value = findIdentity(file, aggregate.schema_name); + if (!value) + return false; + if (value->isNull()) + continue; + + if (aggregate.kind == AggKind::CountColumn) + { + if (common::addOverflow(group.column_counts[agg_index], file.record_count, group.column_counts[agg_index])) + return false; + } + else + { + group.extremal_values[agg_index].push_back(*value); + } + } + } + + Block result; + for (size_t key_index = 0; key_index < group_keys.size(); ++key_index) + { + auto column = group_keys[key_index].type->createColumn(); + for (const auto & group : groups) + column->insert(group.keys[key_index]); + result.insert(ColumnWithTypeAndName(std::move(column), group_keys[key_index].type, group_keys[key_index].action_name)); + } + + for (size_t agg_index = 0; agg_index < aggregates.size(); ++agg_index) + { + const auto & aggregate = aggregates[agg_index]; + auto state_column = ColumnAggregateFunction::create(aggregate.function); + for (const auto & group : groups) + { + Arena arena; + char * place = arena.alignedAlloc(aggregate.function->sizeOfData(), aggregate.function->alignOfData()); + aggregate.function->create(place); + SCOPE_EXIT_MEMORY_SAFE(aggregate.function->destroy(place)); + + if (aggregate.kind == AggKind::CountAll || aggregate.kind == AggKind::CountColumn) + { + const UInt64 count_value = aggregate.kind == AggKind::CountAll ? group.rows : group.column_counts[agg_index]; + /// `count` and `count(column)` both store `AggregateFunctionCountData`. + AggregateFunctionCount::set(place, count_value); + } + else if (!group.extremal_values[agg_index].empty()) + { + auto argument = aggregate.function->getArgumentTypes().at(0)->createColumn(); + for (const auto & value : group.extremal_values[agg_index]) + argument->insert(value); + const IColumn * arguments[] = {argument.get()}; + Arena add_arena; + for (size_t row = 0; row < argument->size(); ++row) + aggregate.function->add(place, arguments, row, &add_arena); + } + state_column->insertFrom(place); + } + result.insert(ColumnWithTypeAndName(std::move(state_column), aggregate.state_type, aggregate.action_name)); + } + + if (settings[Setting::allow_experimental_parallel_reading_from_replicas] > 0 && settings[Setting::max_parallel_replicas] > 1) + { + query_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0)); + LOG_TRACE(getLogger("Planner"), "Disabling parallel replicas to aggregate Iceberg partitions from metadata"); + } + + Chunk chunk(result.getColumns(), result.rows()); + auto chunk_info = std::make_shared(); + chunk.getChunkInfos().add(std::move(chunk_info)); + + auto source = std::make_shared(std::make_shared(result.cloneEmpty()), std::move(chunk)); + auto prepared = std::make_unique(Pipe(std::move(source))); + prepared->setStepDescription("Optimized aggregate by partition columns from metadata"); + query_plan.addStep(std::move(prepared)); + return true; +} + bool applyTrivialCountIfPossible( QueryPlan & query_plan, SelectQueryInfo & select_query_info, @@ -2083,8 +2554,12 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(TableExpressionNodePtr table_ /// Apply trivial_count if possible. The plain variant requires no `WHERE`; the /// sparsity-filter variant requires a `WHERE`, so at most one of them fires. - is_trivial_count_applied = !select_query_options.only_analyze && !select_query_options.build_logical_plan && is_single_table_expression - && (table_node || table_function_node) && select_query_info.has_aggregates + /// Iceberg identity partitions are answered afterwards, so an unfiltered `count()` keeps + /// the existing trivial-count path and its profile event. + const bool can_apply_metadata_aggregate = !select_query_options.only_analyze && !select_query_options.build_logical_plan + && is_single_table_expression && (table_node || table_function_node); + + is_trivial_count_applied = can_apply_metadata_aggregate && select_query_info.has_aggregates && (applyTrivialCountIfPossible( query_plan, table_expression_query_info, @@ -2105,6 +2580,19 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(TableExpressionNodePtr table_ table_expression_data.getColumnNames(), *planner_context)); + if (!is_trivial_count_applied && can_apply_metadata_aggregate + && (select_query_info.has_aggregates || select_query_info.query_tree->as().hasGroupBy())) + { + is_trivial_count_applied = applyIcebergIdentityPartitionAggregationIfPossible( + query_plan, + table_expression_query_info, + table_node, + table_function_node, + select_query_info.query_tree, + planner_context->getMutableQueryContext(), + planner_context); + } + if (is_trivial_count_applied) { till_stage = QueryProcessingStage::WithMergeableState; diff --git a/src/Storages/IStorage.h b/src/Storages/IStorage.h index 95696fdb27b5f..82cac2d1ca308 100644 --- a/src/Storages/IStorage.h +++ b/src/Storages/IStorage.h @@ -25,6 +25,8 @@ #include #include #include +#include +#include namespace DB @@ -81,6 +83,14 @@ using ObjectStoragePtr = std::shared_ptr; class ActionsDAG; +/// One live data file whose identity-partition values and `record_count` are known exactly. +/// A null Field is SQL NULL. A column that is absent is not an identity partition of this file. +struct IdentityPartitionFile +{ + std::vector> identity_values; + UInt64 record_count = 0; +}; + /** Storage. Describes the table. Responsible for * - storage of the table data; * - the definition in which files (or not in files) the data is stored; @@ -870,6 +880,10 @@ It is currently only implemented in StorageObjectStorage. /// Same as above but also take partition predicate into account. virtual std::optional totalRowsByPartitionPredicate(const ActionsDAG &, ContextPtr) const { return {}; } + /// Live data files with identity partition values and `record_count`. + /// nullopt means the storage cannot answer exactly, so the caller must scan. + virtual std::optional> getIdentityPartitionFiles(ContextPtr) const { return {}; } + /// Aggregated `(num_rows, num_defaults)` for `column_name` across all visible parts, /// taken from per-part `SerializationInfo`. Returns nullopt when the storage cannot /// supply an exact count -- see `Storages/MergeTree/SparsityFilter.h` for the precise diff --git a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h index 29ffc5855c280..57f766fcab920 100644 --- a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h +++ b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h @@ -653,6 +653,9 @@ class StorageIcebergConfiguration : public StorageObjectStorageConfiguration, pu bool isStaticConfiguration() const override { return getImpl().isStaticConfiguration(); } bool isDataLakeConfiguration() const override { return getImpl().isDataLakeConfiguration(); } + /// `StorageIcebergConfiguration` selects S3, Azure, HDFS, or local at runtime. + /// `system.iceberg_files` and partition-column aggregates recognize Iceberg through this flag. + bool isIcebergConfiguration() const override { return getImpl().isIcebergConfiguration(); } bool supportsTotalRows(ContextPtr context, ObjectStorageType storage_type) const override { return getImpl().supportsTotalRows(context, storage_type); } std::optional totalRows(ContextPtr context) override { return getImpl().totalRows(context); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 1e4ca36843d3f..b0e824ed003a0 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1630,6 +1630,78 @@ std::optional IcebergMetadata::totalRows(ContextPtr local_context) const return static_cast(result); } +std::optional> IcebergMetadata::getDataFileIdentityPartitions(ContextPtr local_context) const +{ + auto [actual_data_snapshot, actual_table_state_snapshot] = getRelevantState(local_context); + + if (!actual_data_snapshot) + return std::vector{}; + + /// Same fail-closed conditions as `totalRows`: deletes make `record_count` not equal to + /// the number of surviving rows, and an unreadable path must not be answered from metadata. + if (actual_data_snapshot->total_equality_delete_rows.has_value() + && *actual_data_snapshot->total_equality_delete_rows > 0) + return std::nullopt; + + if (!persistent_components.schema_processor) + return std::nullopt; + + std::vector result; + for (const auto & manifest_list_entry : actual_data_snapshot->manifest_list_entries) + { + auto manifest_file_ptr = getManifestFileEntriesHandle( + object_storage, persistent_components, local_context, log, manifest_list_entry, actual_table_state_snapshot.schema_id, *external_storages); + + for (const auto & processed : manifest_file_ptr.getFilesWithoutDeleted(FileContentType::DATA)) + if (!isPathReadable( + persistent_components.table_location, + processed->parsed_entry->file_path_key.serialize(), + object_storage, + local_context)) + return std::nullopt; + + if (!manifest_file_ptr.getFilesWithoutDeleted(FileContentType::EQUALITY_DELETE).empty() + || !manifest_file_ptr.getFilesWithoutDeleted(FileContentType::POSITION_DELETE).empty()) + return std::nullopt; + + for (const auto & processed : manifest_file_ptr.getFilesWithoutDeleted(FileContentType::DATA)) + { + const auto & entry = *processed->parsed_entry; + /// A deletion vector on the data file itself, or a negative count, is not an exact row count. + if (entry.isDeletionVector() || entry.record_count < 0) + return std::nullopt; + /// An empty file contributes no rows and no group. + if (entry.record_count == 0) + continue; + if (!processed->common_partition_specification) + return std::nullopt; + + auto partition_values = Iceberg::getIdentityPartitionColumnValues( + *processed, *persistent_components.schema_processor, /* skip_nulls */ false); + + IdentityPartitionFile file; + file.record_count = static_cast(entry.record_count); + for (auto & [column_name, column_value] : partition_values) + { + auto it = std::find_if(file.identity_values.begin(), file.identity_values.end(), [&](const auto & existing) + { + return existing.first == column_name; + }); + if (it != file.identity_values.end()) + { + if (!(it->second == column_value)) + return std::nullopt; + continue; + } + file.identity_values.emplace_back(std::move(column_name), std::move(column_value)); + } + result.push_back(std::move(file)); + } + } + + return result; +} + std::optional IcebergMetadata::totalBytes(ContextPtr local_context) const { auto [actual_data_snapshot, actual_table_state_snapshot] = getRelevantState(local_context); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index ab1d0762cafec..a93fe24346fe7 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -129,6 +129,11 @@ class IcebergMetadata : public IDataLakeMetadata static bool supportsTotalRows(ContextPtr, ObjectStorageType) { return true; } std::optional totalRows(ContextPtr Local_context) const override; + + /// Live DATA files with identity partition values (including SQL NULL) and `record_count`. + /// nullopt when an exact answer is impossible: equality or position deletes, deletion vectors, + /// an unreadable path, or a negative `record_count`. An empty snapshot is an empty vector. + std::optional> getDataFileIdentityPartitions(ContextPtr local_context) const; static bool supportsTotalBytes(ContextPtr, ObjectStorageType) { return true; } std::optional totalBytes(ContextPtr Local_context) const override; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 3155e86a5c2f9..e6d1c8d668518 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -2187,7 +2187,7 @@ void forEachAvroEntry( } PartitionColumnValues getIdentityPartitionColumnValues( - const ProcessedManifestFileEntry & manifest_file_entry, const IcebergSchemaProcessor & schema_processor) + const ProcessedManifestFileEntry & manifest_file_entry, const IcebergSchemaProcessor & schema_processor, bool skip_nulls) { const auto & partition_key_value = manifest_file_entry.parsed_entry->partition_key_value; if (partition_key_value.empty()) @@ -2228,7 +2228,9 @@ PartitionColumnValues getIdentityPartitionColumnValues( { value = convertFieldToTypeOrThrow(raw_value, *name_and_type->type); } - if (value.isNull()) + /// A null identity value is a real SQL NULL group. Callers that only need the + /// non-null values (pruning, compaction) skip them; aggregation must keep them. + if (value.isNull() && skip_nulls) continue; result.emplace_back(name_and_type->name, std::move(value)); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h index ceda88fc76e12..2280a1b53bd04 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -189,7 +189,9 @@ void forEachAvroEntry( using PartitionColumnValues = std::vector>; PartitionColumnValues getIdentityPartitionColumnValues( - const ProcessedManifestFileEntry & manifest_file_entry, const IcebergSchemaProcessor & schema_processor); + const ProcessedManifestFileEntry & manifest_file_entry, + const IcebergSchemaProcessor & schema_processor, + bool skip_nulls = true); } #endif diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index cb5f3c1abe101..aaf25c83a80d4 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -1,4 +1,8 @@ #include +#include "config.h" +#if USE_AVRO +#include +#endif #include #include @@ -584,6 +588,26 @@ std::optional StorageObjectStorage::totalBytes(ContextPtr query_context) return configuration->totalBytes(query_context); } +std::optional> StorageObjectStorage::getIdentityPartitionFiles(ContextPtr query_context) const +{ +#if USE_AVRO + /// A replica with `distributed_processing` sees only its own tasks. Answering from the whole + /// snapshot here would count every shard's rows on every shard. + if (distributed_processing || !configuration->isIcebergConfiguration()) + return std::nullopt; + + is_table_function ? configuration->lazyInitializeIfNeeded(object_storage, query_context) : configuration->update(object_storage, query_context); + + auto iceberg = std::dynamic_pointer_cast(configuration->getExternalMetadata()); + if (!iceberg) + return std::nullopt; + return iceberg->getDataFileIdentityPartitions(query_context); +#else + (void)query_context; + return std::nullopt; +#endif +} + void StorageObjectStorage::read( QueryPlan & query_plan, const Names & column_names, diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.h b/src/Storages/ObjectStorage/StorageObjectStorage.h index 15c41e5f2e235..f10a3f6f3ea04 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.h +++ b/src/Storages/ObjectStorage/StorageObjectStorage.h @@ -212,6 +212,7 @@ class StorageObjectStorage : public IStorage, public IBackgroundOperation std::optional totalRows(ContextPtr query_context) const override; std::optional totalBytes(ContextPtr query_context) const override; + std::optional> getIdentityPartitionFiles(ContextPtr query_context) const override; bool optimize( const ASTPtr & /*query*/, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 545122e68f0f4..722f32a1467e0 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -331,6 +331,15 @@ std::optional StorageObjectStorageCluster::totalBytes(ContextPtr query_c return configuration->totalBytes(query_context); } +std::optional> StorageObjectStorageCluster::getIdentityPartitionFiles(ContextPtr query_context) const +{ + /// The initiator's `pure_storage` reads the snapshot once. Shards are not asked. + /// `pure_storage` itself returns nullopt when `distributed_processing` is set. + if (pure_storage) + return pure_storage->getIdentityPartitionFiles(query_context); + return std::nullopt; +} + bool StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr & query, ContextPtr context, bool make_cluster_function) { // Change table engine on table function for distributed request diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 9a4a4fc53122e..0da55ea58213d 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -65,6 +65,7 @@ class StorageObjectStorageCluster : public IStorageCluster std::optional totalRows(ContextPtr query_context) const override; std::optional totalBytes(ContextPtr query_context) const override; + std::optional> getIdentityPartitionFiles(ContextPtr query_context) const override; void setClusterNameFromFunctionArgument(bool cluster_name_from_function_argument_) { From d3d59b18f38d50d4c68fca56f3ddcf5b388dc68d Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 26 Oct 2026 18:44:19 +0100 Subject: [PATCH 3/9] Scan unfiltered Iceberg `count` when `optimize_trivial_count_query` is off `SELECT count()` and `count(column)` with no `WHERE` or `GROUP BY` follow the setting. Partition `min`, `max`, and grouped `count` still come from manifests. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 18 +++++++++++++ ...erg_partition_aggregate_metadata.reference | 3 +++ ...28_iceberg_partition_aggregate_metadata.sh | 25 +++++++++++++++++++ 3 files changed, 46 insertions(+) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 44eb9cab3f4a0..727cb8e9c9283 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -936,6 +936,24 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( referenced_columns.insert(referenced_columns.end(), where_columns.begin(), where_columns.end()); } + /// `optimize_trivial_count_query = 0` disables `SELECT count() FROM table`. + /// Trivial count has already declined; do not answer that shape from manifests. + /// `WHERE` and `GROUP BY` are not the trivial-count query, so they stay here. + if (!settings[Setting::optimize_trivial_count_query] && !query_node.hasWhere() && !query_node.hasGroupBy()) + { + bool only_count = !aggregates.empty(); + for (const auto & aggregate : aggregates) + { + if (aggregate.kind != AggKind::CountAll && aggregate.kind != AggKind::CountColumn) + { + only_count = false; + break; + } + } + if (only_count) + return false; + } + auto files = storage->getIdentityPartitionFiles(query_context); if (!files) return false; diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference index 6ecff4cb2cc0b..a081ca6dbb7dc 100644 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -53,6 +53,9 @@ max_of_groupby metadata groupby_d_only metadata count metadata max_k data +count_setting_off +10000 +data icebergLocal files 10 max_d diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index 321ece3c29a3a..c4995a18c24b0 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -183,6 +183,31 @@ exercise() create_table "${LOCAL_TABLE}" "IcebergLocal('${LOCAL_PATH}')" exercise "IcebergLocal" "${LOCAL_TABLE}" + +# `optimize_trivial_count_query = 0` must scan. Manifest `min` / `max` / grouped `count` stay enabled. +echo "count_setting_off" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT count() FROM ${LOCAL_TABLE} + SETTINGS + log_comment = '${PREFIX}_count_setting_off', + log_queries = 1, + optimize_trivial_count_query = 0, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows >= 10000 AND ProfileEvents['IcebergTrivialCountOptimizationApplied'] = 0, + 'data', + concat('metadata read_rows=', toString(read_rows), ' trivial_count=', toString(ProfileEvents['IcebergTrivialCountOptimizationApplied']))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_count_setting_off' + ORDER BY event_time_microseconds DESC + LIMIT 1" + exercise "icebergLocal" "icebergLocal('${LOCAL_PATH}')" create_table "${S3_TABLE}" "IcebergS3(s3_conn, filename = '${S3_KEY}')" From 9a2d4a9fa9a3f7c8d48cdadddc5c0844592790b4 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 26 Oct 2026 18:46:47 +0100 Subject: [PATCH 4/9] Read the latest `query_log` row in `05428` `max(read_rows)` kept a scan from an earlier run with the same `log_comment` and classified the new execution as a data read. Co-authored-by: Cursor --- .../05428_iceberg_partition_aggregate_metadata.sh | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index c4995a18c24b0..eb3da589de13d 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -130,13 +130,16 @@ exercise() ${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" echo "reads" + # `log_comment` is stable across runs in the same database. Take the latest + # `QueryFinish` for each statement. `max(read_rows)` would keep a scan from + # an earlier run and report this run as a data read. ${CLICKHOUSE_CLIENT} --format TabSeparated --query " WITH entries AS ( SELECT replaceOne(log_comment, '${tag}_', '') AS name, - max(read_rows) AS read_rows, - max(ProfileEvents['IcebergTrivialCountOptimizationApplied']) AS trivial_count + argMax(read_rows, event_time_microseconds) AS read_rows, + argMax(ProfileEvents['IcebergTrivialCountOptimizationApplied'], event_time_microseconds) AS trivial_count FROM system.query_log WHERE type = 'QueryFinish' AND is_initial_query From a648b3ee1b37df30bc3b79c12cf0d538a1b4bc3d Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 27 Oct 2026 11:15:59 +0100 Subject: [PATCH 5/9] Read Iceberg partition aggregates from the snapshot pinned at analysis A commit after analysis must not change `min`, `max`, or grouped `count`. The files come from `datalake_table_state`, not the latest metadata. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 5 ++++- src/Storages/IStorage.h | 5 +++-- .../DataLakes/Iceberg/IcebergMetadata.cpp | 12 +++++++++--- .../DataLakes/Iceberg/IcebergMetadata.h | 7 ++++--- src/Storages/ObjectStorage/StorageObjectStorage.cpp | 13 ++++++++++--- src/Storages/ObjectStorage/StorageObjectStorage.h | 2 +- .../ObjectStorage/StorageObjectStorageCluster.cpp | 6 ++++-- .../ObjectStorage/StorageObjectStorageCluster.h | 2 +- 8 files changed, 36 insertions(+), 16 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 727cb8e9c9283..124455c44faf4 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -954,7 +954,10 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( return false; } - auto files = storage->getIdentityPartitionFiles(query_context); + /// Column types come from the storage snapshot pinned at analysis. Read that + /// `datalake_table_state`, not a commit that landed after the pin. + const auto & storage_snapshot = table_node ? table_node->getStorageSnapshot() : table_function_node->getStorageSnapshot(); + auto files = storage->getIdentityPartitionFiles(storage_snapshot->metadata, query_context); if (!files) return false; diff --git a/src/Storages/IStorage.h b/src/Storages/IStorage.h index 82cac2d1ca308..ce8a17a2be886 100644 --- a/src/Storages/IStorage.h +++ b/src/Storages/IStorage.h @@ -881,8 +881,9 @@ It is currently only implemented in StorageObjectStorage. virtual std::optional totalRowsByPartitionPredicate(const ActionsDAG &, ContextPtr) const { return {}; } /// Live data files with identity partition values and `record_count`. - /// nullopt means the storage cannot answer exactly, so the caller must scan. - virtual std::optional> getIdentityPartitionFiles(ContextPtr) const { return {}; } + /// `storage_metadata` is the snapshot pinned at analysis. The files must come from that + /// state, not from a newer commit. nullopt means the storage cannot answer exactly, so the caller must scan. + virtual std::optional> getIdentityPartitionFiles(const StorageMetadataPtr & /*storage_metadata*/, ContextPtr) const { return {}; } /// Aggregated `(num_rows, num_defaults)` for `column_name` across all visible parts, /// taken from per-part `SerializationInfo`. Returns nullopt when the storage cannot diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index b0e824ed003a0..41d0d24ea4ba3 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1630,9 +1630,15 @@ std::optional IcebergMetadata::totalRows(ContextPtr local_context) const return static_cast(result); } -std::optional> IcebergMetadata::getDataFileIdentityPartitions(ContextPtr local_context) const +std::optional> IcebergMetadata::getDataFileIdentityPartitions( + StorageMetadataPtr storage_metadata, ContextPtr local_context) const { - auto [actual_data_snapshot, actual_table_state_snapshot] = getRelevantState(local_context); + /// Same snapshot the scan uses. `getRelevantState` would read the latest metadata file instead. + auto table_state = extractIcebergSnapshotIdFromMetadataObject(storage_metadata); + if (!table_state) + return std::nullopt; + + auto actual_data_snapshot = getRelevantDataSnapshotFromTableStateSnapshot(*table_state, local_context); if (!actual_data_snapshot) return std::vector{}; @@ -1650,7 +1656,7 @@ std::optional> IcebergMetadata::getDataFileId for (const auto & manifest_list_entry : actual_data_snapshot->manifest_list_entries) { auto manifest_file_ptr = getManifestFileEntriesHandle( - object_storage, persistent_components, local_context, log, manifest_list_entry, actual_table_state_snapshot.schema_id, *external_storages); + object_storage, persistent_components, local_context, log, manifest_list_entry, table_state->schema_id, *external_storages); for (const auto & processed : manifest_file_ptr.getFilesWithoutDeleted(FileContentType::DATA)) if (!isPathReadable( diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index a93fe24346fe7..b6d41c7021734 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -131,9 +131,10 @@ class IcebergMetadata : public IDataLakeMetadata std::optional totalRows(ContextPtr Local_context) const override; /// Live DATA files with identity partition values (including SQL NULL) and `record_count`. - /// nullopt when an exact answer is impossible: equality or position deletes, deletion vectors, - /// an unreadable path, or a negative `record_count`. An empty snapshot is an empty vector. - std::optional> getDataFileIdentityPartitions(ContextPtr local_context) const; + /// Resolved from `datalake_table_state` on `storage_metadata`, the snapshot pinned at analysis. + /// nullopt when that state is missing or an exact answer is impossible: equality or position deletes, + /// deletion vectors, an unreadable path, or a negative `record_count`. An empty snapshot is an empty vector. + std::optional> getDataFileIdentityPartitions(StorageMetadataPtr storage_metadata, ContextPtr local_context) const; static bool supportsTotalBytes(ContextPtr, ObjectStorageType) { return true; } std::optional totalBytes(ContextPtr Local_context) const override; diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index aaf25c83a80d4..296eb44c46ce5 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -588,7 +588,8 @@ std::optional StorageObjectStorage::totalBytes(ContextPtr query_context) return configuration->totalBytes(query_context); } -std::optional> StorageObjectStorage::getIdentityPartitionFiles(ContextPtr query_context) const +std::optional> StorageObjectStorage::getIdentityPartitionFiles( + const StorageMetadataPtr & storage_metadata, ContextPtr query_context) const { #if USE_AVRO /// A replica with `distributed_processing` sees only its own tasks. Answering from the whole @@ -596,13 +597,19 @@ std::optional> StorageObjectStorage::getIdent if (distributed_processing || !configuration->isIcebergConfiguration()) return std::nullopt; - is_table_function ? configuration->lazyInitializeIfNeeded(object_storage, query_context) : configuration->update(object_storage, query_context); + /// Analysis pinned `datalake_table_state`. `configuration->update` would load a newer commit + /// and disagree with the column types of this query. A missing pin scans, and the read path pins one snapshot. + if (!storage_metadata || !storage_metadata->datalake_table_state.has_value()) + return std::nullopt; + + configuration->lazyInitializeIfNeeded(object_storage, query_context); auto iceberg = std::dynamic_pointer_cast(configuration->getExternalMetadata()); if (!iceberg) return std::nullopt; - return iceberg->getDataFileIdentityPartitions(query_context); + return iceberg->getDataFileIdentityPartitions(storage_metadata, query_context); #else + (void)storage_metadata; (void)query_context; return std::nullopt; #endif diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.h b/src/Storages/ObjectStorage/StorageObjectStorage.h index f10a3f6f3ea04..1fb41e4de6b4a 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.h +++ b/src/Storages/ObjectStorage/StorageObjectStorage.h @@ -212,7 +212,7 @@ class StorageObjectStorage : public IStorage, public IBackgroundOperation std::optional totalRows(ContextPtr query_context) const override; std::optional totalBytes(ContextPtr query_context) const override; - std::optional> getIdentityPartitionFiles(ContextPtr query_context) const override; + std::optional> getIdentityPartitionFiles(const StorageMetadataPtr & storage_metadata, ContextPtr query_context) const override; bool optimize( const ASTPtr & /*query*/, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 722f32a1467e0..c3f1c0e48cc98 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -331,12 +331,14 @@ std::optional StorageObjectStorageCluster::totalBytes(ContextPtr query_c return configuration->totalBytes(query_context); } -std::optional> StorageObjectStorageCluster::getIdentityPartitionFiles(ContextPtr query_context) const +std::optional> StorageObjectStorageCluster::getIdentityPartitionFiles( + const StorageMetadataPtr & storage_metadata, ContextPtr query_context) const { /// The initiator's `pure_storage` reads the snapshot once. Shards are not asked. /// `pure_storage` itself returns nullopt when `distributed_processing` is set. + /// `storage_metadata` is the cluster snapshot pinned at analysis. `pure_storage` shares this configuration. if (pure_storage) - return pure_storage->getIdentityPartitionFiles(query_context); + return pure_storage->getIdentityPartitionFiles(storage_metadata, query_context); return std::nullopt; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 0da55ea58213d..47cbffc741da1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -65,7 +65,7 @@ class StorageObjectStorageCluster : public IStorageCluster std::optional totalRows(ContextPtr query_context) const override; std::optional totalBytes(ContextPtr query_context) const override; - std::optional> getIdentityPartitionFiles(ContextPtr query_context) const override; + std::optional> getIdentityPartitionFiles(const StorageMetadataPtr & storage_metadata, ContextPtr query_context) const override; void setClusterNameFromFunctionArgument(bool cluster_name_from_function_argument_) { From 8be867529cc31a8b74dbae50c4d87d0a9a24002e Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 27 Oct 2026 11:23:33 +0100 Subject: [PATCH 6/9] Scan Iceberg partition aggregates when `WHERE` uses `rand` A per-row function must not keep a whole file `record_count`. Deterministic functions of the identity column, such as `toYear`, stay on the manifest path. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 9 ++++ ...erg_partition_aggregate_metadata.reference | 5 ++ ...28_iceberg_partition_aggregate_metadata.sh | 50 +++++++++++++++++++ 3 files changed, 64 insertions(+) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 124455c44faf4..8ab844ec428e3 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -821,6 +821,15 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( const auto & function = node->as(); if (!function.isOrdinaryFunction() || function.getFunctionName() == "arrayJoin") return false; + /// One file stands for every row in it. `rand` and `rowNumberInBlock` differ per row, + /// and `sleep` is observable each time it runs. A deterministic function of the identity + /// value, such as `toYear(d)`, is the same for every row in the file. + const auto resolved = function.getFunction(); + if (!resolved + || !resolved->isDeterministicInScopeOfQuery() + || resolved->isStateful() + || resolved->hasObservableSideEffects()) + return false; for (const auto & argument : function.getArguments().getNodes()) { if (!self(argument, out, self)) diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference index a081ca6dbb7dc..1ea9b57e30cb1 100644 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -56,6 +56,11 @@ max_k data count_setting_off 10000 data +rand_filter +data +to_year +10000 +metadata icebergLocal files 10 max_d diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index eb3da589de13d..7bf441b9342c1 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -211,6 +211,56 @@ ${CLICKHOUSE_CLIENT} --format TabSeparated --query " ORDER BY event_time_microseconds DESC LIMIT 1" +# `rand()` differs per row. A match must not keep the whole file `record_count`. +echo "rand_filter" +${CLICKHOUSE_CLIENT} --query " + SELECT count() FROM ${LOCAL_TABLE} + WHERE rand() % 2 = 0 + SETTINGS + log_comment = '${PREFIX}_rand_filter', + log_queries = 1, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" > /dev/null +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows >= 10000, + 'data', + concat('metadata read_rows=', toString(read_rows))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_rand_filter' + ORDER BY event_time_microseconds DESC + LIMIT 1" + +# `toYear(d)` is the same for every row of an identity partition file. +echo "to_year" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT count() FROM ${LOCAL_TABLE} + WHERE toYear(d) = 2026 + SETTINGS + log_comment = '${PREFIX}_to_year', + log_queries = 1, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows <= 10, + 'metadata', + concat('data read_rows=', toString(read_rows))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_to_year' + ORDER BY event_time_microseconds DESC + LIMIT 1" + exercise "icebergLocal" "icebergLocal('${LOCAL_PATH}')" create_table "${S3_TABLE}" "IcebergS3(s3_conn, filename = '${S3_KEY}')" From b92b134d3c391c4704e6b553819ce4dfbf7f2b08 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 28 Oct 2026 09:12:04 +0100 Subject: [PATCH 7/9] Scan Iceberg partition aggregates when `max_rows_to_group_by` is set Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 5 +++ ...erg_partition_aggregate_metadata.reference | 4 +++ ...28_iceberg_partition_aggregate_metadata.sh | 36 +++++++++++++++++++ 3 files changed, 45 insertions(+) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 8ab844ec428e3..93a9bc24fbdca 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -783,6 +783,11 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( || query_node.isGroupByWithGroupingSets()) return false; + /// `max_rows_to_group_by` is enforced while keys are inserted. This path emits every group, + /// and `MergingAggregated` is built without that limit, so `throw`, `break`, and `any` would not apply. + if (query_node.hasGroupBy() && settings[Setting::max_rows_to_group_by]) + return false; + if (!query_node.getCorrelatedColumnsSet().empty() || queryHasArrayJoinInJoinTree(query_tree)) return false; diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference index 1ea9b57e30cb1..8970b3aacfcd7 100644 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -61,6 +61,10 @@ data to_year 10000 metadata +group_by_row_limit +TOO_MANY_ROWS +group_by_row_limit_any +2 icebergLocal files 10 max_d diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index 7bf441b9342c1..38eca059d8ffc 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -261,6 +261,42 @@ ${CLICKHOUSE_CLIENT} --format TabSeparated --query " ORDER BY event_time_microseconds DESC LIMIT 1" +# `max_rows_to_group_by` must use the scan. The manifest path would return every partition. +echo "group_by_row_limit" +set +e +limit_output=$(${CLICKHOUSE_CLIENT} --query " + SELECT d, count() FROM ${LOCAL_TABLE} + GROUP BY d + SETTINGS + max_rows_to_group_by = 1, + group_by_overflow_mode = 'throw', + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" 2>&1) +limit_status=$? +set -e +if [ "${limit_status}" -eq 0 ] || [[ "${limit_output}" != *"TOO_MANY_ROWS"* ]]; then + echo "group_by_row_limit expected TOO_MANY_ROWS, status=${limit_status}" + printf '%s\n' "${limit_output}" + exit 1 +fi +echo "TOO_MANY_ROWS" + +# `any` stops after the key that crosses `max_rows_to_group_by`, so limit 1 keeps two groups. +echo "group_by_row_limit_any" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT count() FROM + ( + SELECT d, count() FROM ${LOCAL_TABLE} + GROUP BY d + SETTINGS + max_rows_to_group_by = 1, + group_by_overflow_mode = 'any', + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0 + )" + exercise "icebergLocal" "icebergLocal('${LOCAL_PATH}')" create_table "${S3_TABLE}" "IcebergS3(s3_conn, filename = '${S3_KEY}')" From 609eba45d440b9d5120597a140e3a5a2699276a6 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 28 Oct 2026 11:48:59 +0100 Subject: [PATCH 8/9] Scan Iceberg partition aggregates when `allow_experimental_iceberg_read_optimization` is off Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 6 +++ ...erg_partition_aggregate_metadata.reference | 5 ++ ...28_iceberg_partition_aggregate_metadata.sh | 50 +++++++++++++++++++ 3 files changed, 61 insertions(+) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 93a9bc24fbdca..e24635ced63ad 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -136,6 +136,7 @@ namespace DB namespace Setting { extern const SettingsMap additional_table_filters; + extern const SettingsBool allow_experimental_iceberg_read_optimization; extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas; extern const SettingsBool optimize_trivial_view_pushdown_to_distributed; extern const SettingsUInt64 distributed_group_by_no_merge; @@ -748,6 +749,11 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( if (settings[Setting::empty_result_for_aggregation_by_empty_set]) return false; + /// `allow_experimental_iceberg_read_optimization` turns off answers taken from Iceberg metadata. + /// `min`, `max`, and filtered or grouped `count` then scan the data files. + if (!settings[Setting::allow_experimental_iceberg_read_optimization]) + return false; + if (query_context->getClientInfo().query_kind == ClientInfo::QueryKind::SECONDARY_QUERY) return false; diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference index 8970b3aacfcd7..9ba22200085c4 100644 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -65,6 +65,11 @@ group_by_row_limit TOO_MANY_ROWS group_by_row_limit_any 2 +read_optimization_off +2026-01-10 +data +read_optimization_off_group +data icebergLocal files 10 max_d diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index 38eca059d8ffc..eaa0b8c77ab1d 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -297,6 +297,56 @@ ${CLICKHOUSE_CLIENT} --format TabSeparated --query " enable_parallel_replicas = 0 )" +# `allow_experimental_iceberg_read_optimization = 0` must scan. Unfiltered `count()` stays on trivial count. +echo "read_optimization_off" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT max(d) FROM ${LOCAL_TABLE} + SETTINGS + log_comment = '${PREFIX}_read_optimization_off', + log_queries = 1, + allow_experimental_iceberg_read_optimization = 0, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows >= 10000, + 'data', + concat('metadata read_rows=', toString(read_rows))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_read_optimization_off' + ORDER BY event_time_microseconds DESC + LIMIT 1" + +echo "read_optimization_off_group" +${CLICKHOUSE_CLIENT} --query " + SELECT d, count() FROM ${LOCAL_TABLE} + GROUP BY d + SETTINGS + log_comment = '${PREFIX}_read_optimization_off_group', + log_queries = 1, + allow_experimental_iceberg_read_optimization = 0, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" > /dev/null +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows >= 10000, + 'data', + concat('metadata read_rows=', toString(read_rows))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_read_optimization_off_group' + ORDER BY event_time_microseconds DESC + LIMIT 1" + exercise "icebergLocal" "icebergLocal('${LOCAL_PATH}')" create_table "${S3_TABLE}" "IcebergS3(s3_conn, filename = '${S3_KEY}')" From 66666d6234bf88c5b6e3de8dde7d2cb97ef6c236 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 29 Oct 2026 13:31:07 +0100 Subject: [PATCH 9/9] Keep Iceberg partition `GROUP BY` on manifests when `max_rows_to_group_by` fits Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 13 +++-- ...erg_partition_aggregate_metadata.reference | 3 ++ ...28_iceberg_partition_aggregate_metadata.sh | 47 ++++++++++++++++--- 3 files changed, 51 insertions(+), 12 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index e24635ced63ad..c876f82b9b6e3 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -789,11 +789,6 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( || query_node.isGroupByWithGroupingSets()) return false; - /// `max_rows_to_group_by` is enforced while keys are inserted. This path emits every group, - /// and `MergingAggregated` is built without that limit, so `throw`, `break`, and `any` would not apply. - if (query_node.hasGroupBy() && settings[Setting::max_rows_to_group_by]) - return false; - if (!query_node.getCorrelatedColumnsSet().empty() || queryHasArrayJoinInJoinTree(query_tree)) return false; @@ -1163,6 +1158,14 @@ bool applyIcebergIdentityPartitionAggregationIfPossible( } } + /// `max_rows_to_group_by` is enforced while keys are inserted, and `MergingAggregated` does not + /// reapply it. The groups fit under the limit for every overflow mode, including the CI profile + /// value. Past the limit, `throw`, `break`, and `any` stay on the scan. + if (!group_keys.empty() + && settings[Setting::max_rows_to_group_by] + && groups.size() > settings[Setting::max_rows_to_group_by]) + return false; + Block result; for (size_t key_index = 0; key_index < group_keys.size(); ++key_index) { diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference index 9ba22200085c4..70f8d15def6fe 100644 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.reference @@ -65,6 +65,9 @@ group_by_row_limit TOO_MANY_ROWS group_by_row_limit_any 2 +group_by_row_limit_fits +10 +metadata read_optimization_off 2026-01-10 data diff --git a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh index eaa0b8c77ab1d..0166b7b8183a0 100755 --- a/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh +++ b/tests/queries/0_stateless/05428_iceberg_partition_aggregate_metadata.sh @@ -283,19 +283,52 @@ fi echo "TOO_MANY_ROWS" # `any` stops after the key that crosses `max_rows_to_group_by`, so limit 1 keeps two groups. +# Each thread applies that on its own, and the merge keeps every partial key, so pin one thread. echo "group_by_row_limit_any" ${CLICKHOUSE_CLIENT} --format TabSeparated --query " SELECT count() FROM ( SELECT d, count() FROM ${LOCAL_TABLE} GROUP BY d - SETTINGS - max_rows_to_group_by = 1, - group_by_overflow_mode = 'any', - optimize_trivial_count_query = 1, - use_iceberg_metadata_files_cache = 0, - enable_parallel_replicas = 0 - )" + ) + SETTINGS + max_rows_to_group_by = 1, + group_by_overflow_mode = 'any', + max_threads = 1, + make_distributed_plan = 0, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" + +# A limit above the partition count still answers from the manifest. The CI profile sets +# `max_rows_to_group_by` to a large non-zero value. +echo "group_by_row_limit_fits" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT count() FROM + ( + SELECT d FROM ${LOCAL_TABLE} + GROUP BY d + ) + SETTINGS + log_comment = '${PREFIX}_group_by_row_limit_fits', + log_queries = 1, + max_rows_to_group_by = 100, + optimize_trivial_count_query = 1, + use_iceberg_metadata_files_cache = 0, + enable_parallel_replicas = 0" +${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS query_log" +${CLICKHOUSE_CLIENT} --format TabSeparated --query " + SELECT if( + read_rows <= 10, + 'metadata', + concat('data read_rows=', toString(read_rows))) + FROM system.query_log + WHERE type = 'QueryFinish' + AND is_initial_query + AND current_database = currentDatabase() + AND log_comment = '${PREFIX}_group_by_row_limit_fits' + ORDER BY event_time_microseconds DESC + LIMIT 1" # `allow_experimental_iceberg_read_optimization = 0` must scan. Unfiltered `count()` stays on trivial count. echo "read_optimization_off"