Skip to content

Fix bucket-merge scheduling and ReadFromCluster filter staleness - #2302

Open
VighneshPath wants to merge 7 commits into
Altinity:antalya-26.6from
VighneshPath:fix/antalya-26.6/query-plan-aggregation-perf
Open

Fix bucket-merge scheduling and ReadFromCluster filter staleness#2302
VighneshPath wants to merge 7 commits into
Altinity:antalya-26.6from
VighneshPath:fix/antalya-26.6/query-plan-aggregation-perf

Conversation

@VighneshPath

@VighneshPath VighneshPath commented Sep 2, 2026

Copy link
Copy Markdown

Found via profiling ClickHouse against StarRocks on the IcebergBench cross-engine benchmark. Two independent performance fixes, bundled on one branch as separate commits (unrelated subsystems, found in the same investigation):

  1. Aggregator::mergeBlocks: two-level aggregation bucket merges were dispatched in ascending bucket-id order. Real bucket-row-count distributions are skewed, so a large bucket landing anywhere in that order left most merge threads idle early while one thread finished it alone — a straggler tail dominating wall time for wide GROUP BY queries. Now dispatches buckets largest-first (longest-processing-time-first scheduling), so the biggest bucket starts while every thread is still free to help, instead of stranding one thread alone with it at the end.

  2. ReadFromCluster (backing icebergCluster(), s3Cluster(), and every other IStorageCluster-based cluster table function): its task/file list was built eagerly from the first applyFilters() call and frozen via a one-shot guard. But query plan optimization calls applyFilters() more than once as the plan is refined — a later pass (e.g. aggregation-in-order for a GROUP BY matching the partition/sort key) can insert another FilterStep above the source, making the filter strictly more complete on a later call. The frozen extension silently dropped anything discovered after the first call. Measured live against Iceberg tables: icebergCluster() read 6-13x more rows than the equivalent ice.\ns.table`query for the identicalWHEREclause, specifically on queries whose plan needed an extra optimization pass.ReadFromObjectStorageStep(the non-cluster object storage path) already gets this right — it defers the equivalent step toinitializePipeline(), which only ever runs after every optimization pass has finished. This change makes ReadFromCluster` follow the same shape.

Both come with unit tests (gtest_aggregator_bucket_merge_order.cpp, gtest_read_from_cluster_predicate_pushdown.cpp) that fail against the pre-fix code and pass against the fix.

Changelog category (leave one):

  • Performance Improvement

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

Improved two-level aggregation merge scheduling to avoid thread-idle tails under skewed bucket sizes, and fixed cluster table functions (icebergCluster(), s3Cluster(), etc.) reading significantly more data than necessary when query optimization refines filter conditions across multiple passes (e.g. GROUP BY on a partitioned column).

Documentation entry for user-facing changes

Not applicable — internal implementation fixes, no new settings or syntax; user-visible effect is fewer files/rows read and less merge-phase tail latency.

CI/CD Options

Exclude tests:

  • Fast test
  • Integration Tests
  • Stateless tests
  • Stateful tests
  • Unit tests
  • Performance tests
  • Aarch64 tests
  • All with ASAN
  • All with TSAN
  • All with MSAN
  • All with UBSAN
  • All with Coverage
  • All Regression
  • Disable CI Cache

Regression jobs to run:

  • Fast suites (mostly <1h)
  • Aggregate Functions (2h)
  • Alter (1.5h)
  • Benchmark (30m)
  • CAS (content-addressed storage; Antalya only)
  • ClickHouse Keeper (1h)
  • Iceberg (2h)
  • LDAP (1h)
  • OAuth (5m)
  • Parquet (1.5h)
  • RBAC (1.5h)
  • SSL Server (1h)
  • S3 (2h)
  • S3 Export (2h)
  • Swarms (30m)
  • Tiered Storage (2h)

VighneshPath and others added 3 commits September 2, 2026 13:52
…atency

Aggregator::mergeBlocks dispatched bucket merges in ascending bucket-id
order. Real bucket-row-count distributions are skewed, so a large bucket
landing anywhere in that order left most merge threads idle early while
one thread finished it alone — a straggler tail dominating wall time for
wide GROUP BY queries (found via flamegraph/profile-event analysis of
IcebergBench's q12_wide_groupby).

Sort buckets by row count descending before dispatch (longest-processing-
time-first), so the biggest buckets start while every thread is still
free to help, and the small ones are left for whoever finishes first.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
ReadFromCluster::createExtension() (backing icebergCluster(), s3Cluster(),
and every other IStorageCluster-based cluster table function) was built
eagerly from applyFilters(), then frozen via a one-shot guard. But query
plan optimizations call applyFilters() more than once as the plan is
refined — a later pass (e.g. aggregation-in-order for a GROUP BY matching
the partition/sort key) can insert another FilterStep above the source,
making filter_actions_dag strictly more complete on a later call. The
frozen extension silently kept whichever predicate happened to be known
on the very first call, dropping anything discovered afterward.

Live testing against Iceberg tables confirmed the effect: for the same
WHERE clause, icebergCluster() read 6-13x more rows than the equivalent
ice.`ns.table` query specifically on GROUP BY-by-partition-column queries
that trigger the extra optimization pass; queries without it were
unaffected, since their filter was already complete on the first call.

ReadFromObjectStorageStep (the non-cluster object storage read path)
already gets this right: it defers building its file iterator to
initializePipeline(), which query optimization only ever reaches after
every pass has finished, so it always sees the final filter. Make
ReadFromCluster follow the same shape: applyFilters() now only updates
filter_actions_dag, and createExtension() is called exactly once, from
initializePipeline(), using the filter's final state.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
ASTPtr is boost::intrusive_ptr<IAST>, not std::shared_ptr, so
std::make_shared<ASTSelectQuery>() didn't convert. And ReadFromCluster's
own applyFilters(ActionDAGNodes) override hides the no-arg
SourceStepWithFilterBase::applyFilters() by name across inheritance
levels when called on the concrete ReadFromCluster type directly, so it
needs explicit base-class qualification.

Caught by the local unit_tests_dbms build (this test was never built
before pushing).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
@VighneshPath
VighneshPath force-pushed the fix/antalya-26.6/query-plan-aggregation-perf branch from 6e2041d to 5671621 Compare September 2, 2026 08:22
VighneshPath and others added 4 commits September 2, 2026 17:02
…rallel replicas

Adds an opt-in `object_storage_cluster_bypass_join_wrap` setting (default off,
behavior unchanged unless a query sets it) that lets a `JOIN` against a
DataLake-catalog-backed table be pushed onto parallel-replica workers together
with partial aggregation, instead of pulling all probe-side rows back to the
initiator and joining there.

Three changes, gated by the setting:
- `PlannerJoinTree.cpp`: skip the leftmost-table subquery-wrap guard that
  normally excludes any multi-table query from an `IStorageCluster`-derived
  source.
- `QueryAnalyzer.cpp` / `TableFunctionsWithClusterAlternativesVisitor.h`: stop
  `parallel_replicas_for_cluster_engines` from being forcibly reset to `false`
  for queries containing a `JOIN`.
- `IStorageCluster.cpp`: serialize the query sent to remote nodes with
  `queryNodeToDistributedSelectQuery()` instead of `query_info.query`, so a
  `JOIN`'s right-hand side referencing a CTE (e.g. `WITH x AS (...) ... JOIN x`)
  is inlined by body rather than sent as a bare, unresolvable CTE name.

Also adds diagnostic `LOG_WARNING` tracing (`DLPR`, `OSC_STAGE`, `CLUSTER_ALT`)
in `DatabaseDataLake.cpp`, `StorageObjectStorageCluster.cpp`, and
`QueryAnalyzer.cpp` to make the parallel-replicas eligibility decision
observable end-to-end.

Validated so far on q17 of an internal Iceberg benchmark: correct results
(checksum match), JOIN + partial aggregation confirmed pushed onto workers via
`system.query_log` `JoinProbeTableRowCount`, ~37% faster hot runtime. The CTE
fix (this commit's `IStorageCluster.cpp` change) is not yet re-validated
end-to-end against the live cluster. See ICEBERG_JOIN_EXPERIMENT.md for full
findings, dead ends, and open items — none of this is production-ready
(diagnostic logging left in, no auto-detection of when the bypass is actually
safe, setting is a blanket per-query opt-in rather than a narrow condition).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…replacement

Redesigns the q21 whole-query dispatch (nested IStorageCluster driver below a
CTE, e.g. `transaction_event` -> `txnlog`) on `StorageDistributed`'s
`cloneAndReplace()` pattern instead of the previous AST-position/name-search
approach: `StorageObjectStorageCluster::buildClusterTableFunctionAST()` builds
the driver's `icebergS3Cluster(...)` replacement by reusing the existing,
already-proven leftmost-table rewrite machinery; `buildQueryPlanForObjectStorageCluster()`
swaps it in via `IQueryTreeNode::cloneAndReplace()` (exact node identity, no
AST search, correct for self-joins by construction); `IStorageCluster::readPreparedClusterQuery()`
dispatches the already-prepared query directly.

Found and fixed four live bugs surfaced by testing against the real cluster:
a `query_info.table_expression`/`planner_context` identity mismatch after
`cloneAndReplace()`; this dispatch representing a whole remote query rather
than a single-table read (no per-table planner-context lookup applies); the
leftmost-table JOIN-wrap bypass being unconditional instead of gated on
whether the JOIN's other side is safely serializable by the legacy
`IStorageCluster::read()` path, which broke `q2`/`q4`/`q13`/`q16` (CTE on the
RHS); and `findObjectStorageClusterWholeQueryDriver()` wrongly treating a
plain subquery chain with no JOIN at all as a whole-query candidate.

Full 23-query IcebergBench suite now passes with correct checksums, and
`q17`/`q21` show the intended parallel-replicas JOIN pushdown speedup vs. the
published baseline. Not yet resolved: a new hot-latency regression on
`q2`/`q4`/`q13` (not `q4`'s sibling `q16`), suspected to be a side effect of
the experimental setting also loosening an unrelated, pre-existing analyzer
check for tables that don't benefit from cluster distribution (e.g. a
window-function CTE that must fully materialize on one node regardless) --
not yet confirmed with log evidence. See ICEBERG_JOIN_EXPERIMENT.md §3.7.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…l-replicas candidate machinery

Replaces the prior ad hoc, three-gate `object_storage_cluster_bypass_join_wrap`
dispatch (spread across `QueryAnalyzer.cpp`/`PlannerJoinTree.cpp`/
`findParallelReplicasQuery.cpp`) with a single pluggable eligibility predicate,
`isObjectStorageClusterDriverEligible()`, reused by both of ClickHouse's own
stock parallel-replicas driver-selection mechanisms:

- `findQueryForParallelReplicas()`/`findTableForParallelReplicas()` (a driver
  nested under a CTE, e.g. `q21`), generalized with a `relax_for_object_storage_driver`
  policy so a JOIN's non-driver branch needing its own finalization doesn't
  wrongly narrow the candidate the way it must for `MergeTree`.
- `allowParallelReplicasForJoinTree()` (a driver as the immediate leftmost table
  of a JOIN, e.g. `q17`), now parameterized over a pluggable
  `ParallelReplicasStorageEligibility` instead of being `MergeTree`-only.

Both dispatch to one shared execution backend, `buildQueryPlanForObjectStorageCluster()`,
mirroring `StorageDistributed::buildQueryTreeDistributed()`'s exact-`QueryTree`-node
replacement pattern.

Also fixes a `StorageDummy`-erases-type bug (the disposable dummy-plan-walk used to
evaluate a candidate's shape substitutes every table with a generic `StorageDummy`,
which silently failed the object-storage eligibility check for a driver reachable
only through a CTE), and adds a `!select_query_options.is_subquery` capability
boundary to the leftmost-driver mechanism: whole-query dispatch is only proven
correct when the JOIN's owning query is the true top level, not nested under
further derived-table wrapping (found live via IcebergBench `q16`, whose JOIN sits
four subquery layers below a Looker-style window-function pivot).

Reverts `QueryAnalyzer.cpp`/`TableFunctionsWithClusterAlternativesVisitor.h` to
stock: cluster dispatch is now injected explicitly, per selected driver, at
execution time, so no query-scope-wide analyzer flag is needed anymore.

See ICEBERG_JOIN_EXPERIMENT.md §12 for the full investigation, including four
other live bugs found and fixed along the way (unbounded `only_analyze`
recursion, an `isDistributed()` misuse, and two gaps in the branch-role-aware
candidate-narrowing walk).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
@ianton-ru

Copy link
Copy Markdown

Partially crossed with #2249

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants