Add object_storage_cluster_join_mode='distributed' for whole-query … - #2383
Open
VighneshPath wants to merge 20 commits into
Open
VighneshPath wants to merge 20 commits into
VighneshPath wants to merge 20 commits into
Conversation
…JOIN dispatch When a JOIN's driving table is a `DataLake` catalog table distributed via `object_storage_cluster`, send the whole query to that table's cluster and merge the partial aggregates on the initiator, instead of reading every table back to the initiator and joining there. The JOIN and any `GROUP BY` then run on every node of the cluster rather than on one. `findDistributedObjectStorageCandidate` decides eligibility. It walks down the left side of the join tree to find the driving table, passing through a subquery only when that subquery does not itself aggregate, deduplicate, sort or limit -- each worker runs it against its own slice of the driver, so anything that finalizes across rows would turn a partial result into a final one. Every other table reachable in the query must resolve through a `DataLake` catalog, carry no row policy, and be readable by the current user. `buildDistributedObjectStorageQueryPlan` then replaces the driver with an explicit `*Cluster()` table function, keyed on the exact query tree node, and reads the result back at `WithMergeableState` through a single `ReadFromCluster` step, so the planner's ordinary finalization applies on top. It follows `buildQueryPlanForParallelReplicas` step for step, including the position-based conversion from the rewritten query's header back to the original's. Only the driving table is partitioned across the cluster; every other table in the query is read and recomputed in full on each node. That cost is not estimated when deciding to dispatch. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
Reuse mechanisms that already exist rather than adding parallel ones: `DatabaseCatalog::isDatalakeCatalog` in place of a per-storage marker, `StorageObjectStorageCluster::getClusterName` in place of a second worker-localization path in `DatabaseDataLake`, and `SourceStepWithFilterBase::applyFilters` in place of clearing fields on the `SelectQueryInfo` handed to `ReadFromCluster`. `DatabaseDataLake.cpp` is no longer touched by this feature at all. The setting's documentation described a search for the highest eligible enclosing query, and a fallback for the level an ineligible table appears at. Neither exists: dispatch is attempted only for the outermost `SELECT` of an initial query, and is all-or-nothing. It now says so, and states that only the driving table is partitioned while the rest is recomputed in full per node. `allWorkerLocalReferencesAreSafe` is renamed to `allWorkerLocalTableReferencesAreSafe`, because it proves nothing about ordinary functions: `dictGet`, a user-defined function or `hostName` move from the initiator to the workers unexamined, as they do for `Distributed`. Tests cover both shapes against a real `DataLake` catalog -- the driver as the JOIN's leftmost table, and the driver behind a subquery with the aggregation on the enclosing query and a nested JOIN on the right -- in derived-table and CTE spellings, comparing results against `object_storage_cluster_join_mode='allow'`. They assert that the whole query reaches a worker, that only the driver becomes a cluster function, and that no partner table fans out again from a worker. A local `Memory` JOIN partner must fall back to ordinary planning. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
…cluster-distributed-join Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
VighneshPath
force-pushed
the
feature/antalya-26.6/object-storage-cluster-distributed-join
branch
from
September 16, 2026 10:06
037f342 to
10f70a2
Compare
…cluster-distributed-join
Member
…cluster-distributed-join
30 tasks
Under `object_storage_cluster_join_mode='distributed'` a worker resolves every DataLake-catalog table through `DatabaseDataLake::tryGetTableImpl`, which builds a fresh `StorageObjectStorageCluster` from the query context. Its constructor decides there and then whether the inner plain storage consumes the initiator's file-task queue, from `collaborate_with_initiator` and the parallel-replicas settings alone -- it does not consider which table this is. Under whole-query dispatch the driver owns that queue, so a partner table answering yes as well would read the driver's files under its own schema. Nothing on the path `getClusterName` -> `readFallBackToPure` -> `StorageObjectStorage::read` -> `createFileIterator` -> `ReadTaskIterator` re-checks this, and `ReadTaskIterator` takes whatever the callback returns without filtering by table. The new test pins the combination of settings that makes the constructor's condition true for every catalog table in the query, and checks the result against ordinary planning. It also asserts the query was still dispatched, so it cannot silently stop covering anything if the candidate is rejected whenever those settings are set. Not yet executed: this machine's container runtime is podman, and `compose/docker_compose_keeper.yml` relies on shell-style default expansion in `entrypoint` that podman-compose does not implement. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
The previous commit described this test as covering a suspected defect. Running it against a local two-node cluster with a real Iceberg REST catalog shows there is no defect: the partner table is constructed on the worker but never consumes the driver's file-task queue, and `ReadTaskIterator` does not appear in the worker log at all. Results match ordinary planning in every combination tried, including with `object_storage_cluster` set as a `DatabaseDataLake` database setting. Two guards are responsible, and neither was traced fully when the concern was raised. `DatabaseDataLake::tryGetTableImpl` only falls back to the parallel-replicas cluster when `!is_secondary_query`, so on a worker the cluster name stays empty and `can_use_parallel_replicas` is false. A dispatched worker query also contains a `*Cluster` table function, which makes the context distributed and fails the `!isDistributed` term. The test stays, as a guard rather than a reproducer: both conditions are incidental to this feature, and removing either would silently turn a partner into a queue consumer reading the driver's files. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
…tion `object_storage_cluster_join_mode='distributed'` identified the driving table to a worker structurally: the planner rewrote that one table expression into an explicit `icebergS3Cluster(...)` call, which the worker then resolved to a task-consuming storage. That worked, but it is why the dispatch path had to fabricate a throwaway SELECT to extract a table function from, re-run `QueryAnalysisPass` on the synthesized node, `cloneAndReplace` it into a copy of the tree, and reconcile two headers built from two different trees. The structural encoding was needed because no name survives the trip. The analyzer assigns every table expression a fresh `__tableN` alias in `createUniqueAliasesIfNecessary`, overwriting whatever the SQL carried, and `queryNodeToDistributedSelectQuery` inlines CTE bodies, so neither an alias nor a position is stable from initiator to worker. A `StorageID` is stable. The query now crosses the wire exactly as written, and the driving table is named alongside it in `object_storage_distributed_driver_database` and `object_storage_distributed_driver_table`. A worker reads that one table from the initiator's file-task queue and every other table in full, locally. That name has to be unambiguous, so the planner counts the driver's occurrences in the serialized query -- after CTE inlining, which is where a single tree node can become two table references -- and declines to dispatch unless it appears exactly once. A self-join or a twice-referenced CTE over the driver falls back to ordinary planning rather than risk two readers of one queue, which would be silently wrong. On the worker side this replaces an inference with a fact. `getClusterName` keyed off `SECONDARY_QUERY && collaborate_with_initiator && hasClusterFunctionReadTaskCallback() && join_mode == DISTRIBUTED` -- none of which is unique to this feature -- and now keys off the announcement. `distributed_processing` stays a construction-time property, so `totalRows`, `totalBytes` and `getPathSample` continue to agree with it; `DatabaseDataLake` builds a fresh storage per resolution, so the query context is available where the decision is made. Verified on a two-node cluster against an Iceberg REST catalog. With a six-file driver and a three-row partner, each node reads nine driver rows and the whole partner, and the merged result matches `join_mode='allow'` exactly. A self-join declines dispatch and returns the same answer through ordinary planning. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
`ReadFromCluster` reads one table: it carries that table's `StorageSnapshot`, its required columns, and a `SelectQueryInfo` describing its table expression, and it pushes filters into all three. Whole-query dispatch reused it with an `is_whole_query_dispatch` flag, which left the same members meaning two different things -- for a dispatched query the snapshot and columns describe the driver while the step's output is the query's result. Each place that noticed needed a branch: `applyFilters` had to fall back to the base implementation because the per-table column mapping throws, `createExtension` had to suppress the filter because it is not a predicate over the driver, and `updateSettings` had to scrub a setting only in one mode. `ReadFromClusterQuery` holds only what dispatching a query needs: the query, the cluster, the output header, and the storage that hands out file tasks. It derives from `ISourceStep`, so there is no filter-pushdown surface to special-case and no `required_source_columns` to disagree with the output. The three branches and the flag are gone rather than better documented. The transport both steps share -- replicas as shards, one connection each, all pulling from one task iterator -- is extracted verbatim into `buildClusterFunctionRemotePipe`. Nothing about it changes. Also removes `buildClusterTableFunctionAST`, unused since the driver stopped being rewritten. With it goes the throwaway `SELECT` it built to run the per-engine rewrite against, and `extractTableFunctionFromSelectQuery`'s last caller on this path. The gtest asserting `required_source_columns` are the driver's columns is replaced rather than fixed: the mismatch it guarded is now unrepresentable, so it instead pins that dispatch produces `ReadFromClusterQuery` and no `ReadFromCluster`. Driver-only file pruning is still not recovered -- `getTaskIteratorExtension` is called with no predicate, so every file of the driver is listed. That needs the conjuncts whose columns all come from the driver, and is now a self-contained change in one place; the comment there says so. Verified on the two-node cluster: same results as `join_mode='allow'`, driver files still split evenly, and `EXPLAIN` shows `ReadFromClusterQuery` under `MergingAggregated`. 41/41 unit tests pass. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
`QueryConditionCache` is keyed by `(table uuid, part name, condition hash)`. For `MergeTree` the part name is unique within a table, so passing a name is enough. Object storage paths are hierarchical, and `StorageObjectStorageSource` was passing `ObjectInfo::getFileName`, which is only the last component. A Hive-partitioned dataset repeats the same file name in every partition -- `day=2025-02-05/part-00000.parquet` and `day=2025-02-27/part-00000.parquet` -- so all of them shared a single cache entry. The first partition whose row groups all failed the condition wrote "nothing matches" under that shared key, and every later query read it back for its namesakes and skipped them without opening them. The result was a table that answered correctly once and returned no rows from those files afterwards, with no error and nothing logged above debug level. Only reachable when the table id is stable across queries, so a table (or a data lake catalog table) hits it while a table function does not: a table function gets a fresh id per query, and its entries never collide. Pass `ObjectInfo::getPath` on both the read and the write side instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A data lake catalog table is constructed with `lazy_init`, so its configuration -- and with it the path the listing walks -- is resolved on first use rather than at construction. Ordinary reads reach the listing through `StorageObjectStorage`, which does that resolution itself. Whole-query dispatch calls `getTaskIteratorExtension` directly and skips it, so a driver that nothing else had touched yet would list no files, hand every worker an empty queue, and return no rows without an error. `lazyInitializeIfNeeded` is a no-op once the configuration is resolved, so this is safe on the paths that already did it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…cluster-distributed-join
Inherited from `buildQueryPlanForParallelReplicas`, which needs it: that one rewrites the table expression before serializing the query, so the header it computes and the header the remote plan produces come from different trees and can disagree. Whole-query dispatch sends the query as written, and `readPreparedClusterQuery` is given `remote_header` as the source step's own output header, so the plan's current header is already that block and the conversion is an identity. It is not free: it puts an `ExpressionStep` over blocks carrying aggregate state at `WithMergeableState` for no benefit. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`object_storage_distributed_driver_database` and `_table` name the one table whose files the initiator hands out. Whole-query dispatch sets them on its own context, from its own step. Nothing marks them internal -- ClickHouse has no such tier -- so a user can set them by hand on any query. On an ordinary cluster read that is mostly harmless, because they are only read on a worker, and a worker only reaches that state when the initiator supplied a task iterator. `ReadFromCluster` does supply one: under parallel replicas its workers see `collaborate_with_initiator`, so a name that reached them would make the table so named read this step's file-task queue instead of listing its own files. Clear both in `ReadFromCluster::updateSettings`, and again from the query's own `SETTINGS` clause, which a worker applies on top of the context settings and would otherwise use to restore them. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…-join' from fork Picks up the antalya-26.6 merge pushed to the branch (CI scripts and test compose files only; no change to the dispatch code). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`ReadFromCluster` cleared the driver announcement from both the context and the query text, but whole-query dispatch cleared neither from the text. Its own announcement travels in the context, and a worker applies the query's `SETTINGS` clause on top of that -- so a name written into the query by hand won the tie and the initiator's choice was overridden. The table the user named, not the one the planner picked, would then read the file-task queue, silently returning wrong rows rather than failing. The three helpers that did this were applied ad hoc per path, which is what made it easy to give one path a subset of the other's. Replace them with one function per kind of remote read, named for that kind, so the set of normalizations is chosen once: - `prepareOrdinaryClusterQueryForRemoteExecution` -- downgrade the join mode, drop the driver name - `prepareWholeQueryDispatchForRemoteExecution` -- drop the cluster, drop the driver name Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`object_storage_distributed_driver_database` and `_table` tell a worker which table reads from the initiator's file-task queue. A server that does not know a setting ignores it with a warning unless it is `IMPORTANT`, so during a rolling upgrade a new initiator could dispatch to an older worker that silently dropped the announcement, found no driver, read every file of every table, and returned each row once per node. `IMPORTANT` makes that worker refuse the query instead. The tier bits and the `IMPORTANT` bit are disjoint (`BaseSettingsHelpers::getTier` masks with `Flags::TIER`), so the settings stay `EXPERIMENTAL`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two mechanisms can put a storage on an initiator's file-task queue: being the announced driver of a whole-query dispatch, or being an ordinary cluster read under parallel replicas. They were combined with `||`, so both could answer yes for different tables in the same query -- and there is one queue, holding one table's files. A partner that qualified for parallel replicas would read the driver's files under its own schema. The reachability of that depended on conditions no invariant pinned: the database's own `object_storage_cluster` leaving `cluster_name_` non-empty on a worker, plus the parallel-replica settings. Rather than rely on those not lining up, make the dispatch decide alone while it is in effect: its announced driver consumes the queue, every other table in that query lists its own files. This is the shape parallel replicas already uses for the same hazard -- one table is chosen, and the mechanism is explicitly disabled for every other (`PlannerJoinTree.cpp`, `parallel_replicas_table`). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The integration test still asserted that a dispatched query contains `icebergS3Cluster(...)` -- the previous design, where the driver was rewritten into a cluster table function. The current one sends the driver as the catalog table the user wrote and names it in the settings, so that assertion could only fail. It was not caught because the branch's `amd_debug` build died in Checkout Submodules and the integration jobs never ran. Rewrite `_assert_dispatched_whole` around what is now true: a worker received the whole query carrying the announcement for the expected driver, no query contains a cluster table function, and no secondary query lacks the JOIN (which is what a table fanning out again from a worker would look like). In the unit tests the same check was made by proxy -- a fake table function was registered so the plan text could be searched for it. Replace that with a tripwire on the rewrite hook itself, which is both the thing being asserted and enough to delete `FakeDriverTableFunction`, its factory registration and its AST surgery. It is reset per plan so it keeps meaning something for the tests that fall back to an ordinary cluster read, where calling the hook is correct. Also corrects comments in the candidate finder and its test that still described the driver as being rewritten. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The dispatch handed out every file of the driver. The step's own filter is no help -- its output is the whole query's result, so a predicate over that says nothing about which files are needed -- and nothing went looking for the driver's predicate elsewhere, so a query selecting one day out of sixty still distributed all sixty days' files and had each worker open its share only to discard them. The predicate is in the query tree, in the `WHERE` of the query nodes crossed on the way down to the driver. Collect those, split them into `and` atoms, keep the atoms whose every column comes from the driver, and build an `ActionsDAG` from them with `buildFilterInfo` -- the same helper row policies and additional table filters use. An atom that reads a partner's column, or contains a subquery, is left alone and simply does not prune. Safe because of two restrictions the candidate finder already enforces: the driver sits on the left spine of INNER ALL / LEFT joins only, so a driver row dropped here cannot have produced a result row; and every query node crossed to reach it is partition-preserving, so dropping a row cannot change what the surviving rows compute. Relaxing either would make this unsound, and the comment says so. Measured on a 600-file, 60-day-partitioned driver with a one-day window: 600 file tasks before, 10 after, with the result unchanged. A query with no driver predicate, and one whose only predicate mixes driver and partner columns, both still distribute all 600. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
…JOIN dispatch
Introduces whole-query dispatch for
JOINs involving aDataLake-catalog table backed byobject_storage_cluster:findDistributedObjectStorageCandidate()walks the left spine from the root query to find such a driver table where every other table reachable in the sameJOIN/subquery tree also safely resolves through aDataLakecatalog, andbuildDistributedObjectStorageQueryPlan()rewrites the driver's table expression into a cluster table function viaIStorageCluster::buildClusterTableFunctionAST()(usingQueryTree::cloneAndReplace(), not in-place mutation) and dispatches the whole enclosing query to the driver's cluster nodes.IStorageCluster::readPreparedClusterQuery()andReadFromClustergained anis_whole_query_dispatchmode: driver-side filter/task-iterator pruning is suppressed (safe fallback, since the shipped query already carries its ownWHERE), andquery_info'stable_expression/planner_context-- which describe a single table for an ordinary per-table read -- are reset beforeReadFromClusteris constructed, since they otherwise get consulted byReadFromCluster::applyFilters()as if they described the whole dispatched query, corrupting filter pushdown and, if the resulting exception path is ever hit, dereferencing a nulltable_expression.Verified against a live Iceberg/MinIO-backed cluster with q17- and q21-shaped
JOINqueries: correct, disjoint fan-out results across cluster nodes, filter pushdown intoReadFromClustersurvives real query plan optimization, no crash.Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
...
Documentation entry for user-facing changes
...
CI/CD Options
Exclude tests:
Regression jobs to run: