Add object_storage_cluster_join_mode='distributed' for whole-query …

src/Core/Settings.cpp

@@ -2138,6 +2138,15 @@ Possible values:
 - `local` — Replaces the database and table in the subquery with local ones for the destination server (shard), leaving the normal `IN`/`JOIN.`
 - `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN.` Right table executes first and is added to the secondary query as temporay table.
 - `allow` — Default value. Allows the use of these types of subqueries.
+- `distributed` — Experimental. Dispatches a whole `JOIN` query to the cluster of its driving table, so the `JOIN` and any `GROUP BY` run on the cluster's nodes instead of on the initiator, which then only merges the partial aggregate states. The driving table is found by walking down the left side of the query -- through `INNER ALL`/`LEFT JOIN`s and through a subquery or CTE that does not itself aggregate, deduplicate, sort or limit -- and must be a `DataLake` catalog table distributed via `object_storage_cluster`.
+
+  This is attempted only for the outermost `SELECT` of an initial query, and it is all-or-nothing: either the entire query is eligible and is dispatched as one unit, or ordinary planning handles the entire query. No narrower, nested candidate is attempted, and nothing falls back per level.
+
+  Eligibility requires that every other table reachable anywhere in the query also resolves through a `DataLake` catalog, has no row-level security policy, and is readable by the current user. An explicit `*Cluster()` table function, an ordinary local or `Distributed` table, or any table failing one of those checks makes the whole query ineligible. The combination is disabled outright when `additional_table_filters` is set or `object_storage_remote_initiator` is enabled.
+
+  Performance characteristic to be aware of: only the driving table is partitioned across the cluster. Every other table in the query is read and recomputed **in full on every node**, including any `JOIN` or `GROUP BY` over them. That is a win when those relations are small relative to the driver and a loss when they are not; ClickHouse does not estimate this cost when deciding to dispatch.
+
+  ClickHouse also does not verify that the tables, dictionaries or user-defined functions the query references are configured identically on every node of the driver's cluster -- as with `Distributed`, that consistency is the deployment's responsibility.
 )", 0) \
     \
     DECLARE(UInt64, max_concurrent_queries_for_all_users, 0, R"(
@@ -8348,6 +8357,18 @@ Trigger processor to spill data into external storage adpatively. grace join is
     DECLARE(String, object_storage_cluster, "", R"(
 Cluster to make distributed requests to object storages with alternative syntax.
 )", EXPERIMENTAL) \
+    DECLARE(String, object_storage_distributed_driver_database, "", R"(
+Internal. Set by the initiator on a query dispatched by `object_storage_cluster_join_mode='distributed'`, naming the
+database of the one table whose files are distributed across the cluster. Together with
+`object_storage_distributed_driver_table` it tells a worker which table reads from the initiator's file-task queue;
+every other table in the same query is read in full, locally. Not meant to be set by hand.
+
+`IMPORTANT` because a worker that silently ignored it would find no driver, read every file of every table, and
+return each row once per node. A server too old to know the setting must refuse the query instead.
+)", EXPERIMENTAL | IMPORTANT) \
+    DECLARE(String, object_storage_distributed_driver_table, "", R"(
+Internal. The table name counterpart of `object_storage_distributed_driver_database`. Not meant to be set by hand.
+)", EXPERIMENTAL | IMPORTANT) \
     DECLARE(UInt64, object_storage_max_nodes, 0, R"(
 Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc.
 Possible values:

src/Core/SettingsChangesHistory.cpp

@@ -42,6 +42,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},
+            {"object_storage_distributed_driver_database", "", "", "New internal setting. Names the database of the driving table of an `object_storage_cluster_join_mode='distributed'` dispatch."},
+            {"object_storage_distributed_driver_table", "", "", "New internal setting. Names the driving table of an `object_storage_cluster_join_mode='distributed'` dispatch."},
         });
 
         addSettingsChanges(settings_changes_history, "26.6",

src/Core/SettingsEnums.cpp

@@ -100,9 +100,10 @@ IMPLEMENT_SETTING_ENUM(DistributedProductMode, ErrorCodes::UNKNOWN_DISTRIBUTED_P
      {"allow",  DistributedProductMode::ALLOW}})
 
 IMPLEMENT_SETTING_ENUM(ObjectStorageClusterJoinMode, ErrorCodes::BAD_ARGUMENTS,
-    {{"local",  ObjectStorageClusterJoinMode::LOCAL},
-     {"global", ObjectStorageClusterJoinMode::GLOBAL},
-     {"allow",  ObjectStorageClusterJoinMode::ALLOW}})
+    {{"local",       ObjectStorageClusterJoinMode::LOCAL},
+     {"global",      ObjectStorageClusterJoinMode::GLOBAL},
+     {"allow",       ObjectStorageClusterJoinMode::ALLOW},
+     {"distributed", ObjectStorageClusterJoinMode::DISTRIBUTED}})
 
 
 IMPLEMENT_SETTING_ENUM(QueryResultCacheNondeterministicFunctionHandling, ErrorCodes::BAD_ARGUMENTS,

src/Core/SettingsEnums.h

@@ -170,7 +170,8 @@ enum class ObjectStorageClusterJoinMode : uint8_t
 {
     LOCAL,       /// Convert to local query
     GLOBAL,      /// Convert to global query
-    ALLOW        /// Enable
+    ALLOW,       /// Enable
+    DISTRIBUTED  /// Let a distributed object-storage driver own the whole JOIN and run it on cluster workers
 };
 
 DECLARE_SETTING_ENUM(ObjectStorageClusterJoinMode)

src/Planner/Planner.cpp

@@ -82,6 +82,8 @@
 #include <Planner/CollectSets.h>
 #include <Planner/CollectTableExpressionData.h>
 #include <Planner/findQueryForParallelReplicas.h>
+#include <Planner/findDistributedObjectStorageCandidate.h>
+#include <Planner/buildDistributedObjectStorageQueryPlan.h>
 #include <Planner/PlannerActionsVisitor.h>
 #include <Planner/PlannerContext.h>
 #include <Planner/PlannerCorrelatedSubqueries.h>
@@ -2302,7 +2304,31 @@ void Planner::buildPlanForQueryNode()
     }
 
     JoinTreeQueryPlan join_tree_query_plan;
-    if (planner_context->getMutableQueryContext()->canUseTaskBasedParallelReplicas()
+    /// object_storage_cluster_join_mode='distributed': only the outermost, initial-query Planner may dispatch.
+    /// All three guards below are load-bearing; each was added after a live failure:
+    ///   - only_analyze: buildDistributedObjectStorageQueryPlan calls getSampleBlock, which starts its own
+    ///     analyze-only Planner over the same query. Without this guard that Planner dispatches again, and so on
+    ///     -- `Code: 306. TOO_DEEP_RECURSION`.
+    ///   - is_subquery: a nested query dispatched on its own gets its `__tableN` identifiers renumbered locally,
+    ///     which no longer match what the enclosing scope resolved against -- `Not found column
+    ///     __table7.appinfo_ccl in block. There are only columns: __table2.appinfo_ccl, ...`.
+    ///   - INITIAL_QUERY: a worker must execute what it was sent, not dispatch it onwards.
+    /// Whole-or-nothing for the outermost query: on rejection, ordinary planning handles everything.
+    std::optional<DistributedObjectStorageCandidate> distributed_object_storage_candidate;
+    if (!select_query_options.only_analyze && !select_query_options.is_subquery
+        && query_context->getClientInfo().query_kind == ClientInfo::QueryKind::INITIAL_QUERY)
+        distributed_object_storage_candidate = findDistributedObjectStorageCandidate(query_tree, query_context);
+
+    /// Dispatch can still decline here: the driver has to be nameable unambiguously in the serialized query.
+    std::optional<JoinTreeQueryPlan> dispatched_query_plan;
+    if (distributed_object_storage_candidate)
+        dispatched_query_plan = buildDistributedObjectStorageQueryPlan(query_tree, *distributed_object_storage_candidate, select_query_info, planner_context);
+
+    if (dispatched_query_plan)
+    {
+        join_tree_query_plan = std::move(*dispatched_query_plan);
+    }
+    else if (planner_context->getMutableQueryContext()->canUseTaskBasedParallelReplicas()
         && planner_context->getGlobalPlannerContext()->parallel_replicas_node == &query_node)
     {
         join_tree_query_plan = buildQueryPlanForParallelReplicas(query_node, planner_context, select_query_info.storage_limits);

src/Planner/buildDistributedObjectStorageQueryPlan.cpp

@@ -0,0 +1,215 @@
+#include <Planner/buildDistributedObjectStorageQueryPlan.h>
+
+#include <Analyzer/ColumnNode.h>
+#include <Analyzer/FunctionNode.h>
+#include <Analyzer/QueryNode.h>
+#include <Analyzer/TableNode.h>
+#include <Analyzer/UnionNode.h>
+#include <Common/Exception.h>
+#include <Interpreters/ActionsDAG.h>
+#include <Interpreters/Context.h>
+#include <Interpreters/InterpreterSelectQueryAnalyzer.h>
+#include <Interpreters/SelectQueryOptions.h>
+#include <Parsers/ASTIdentifier.h>
+#include <Parsers/ASTSelectQuery.h>
+#include <Planner/PlannerContext.h>
+#include <Planner/Utils.h>
+#include <Storages/IStorageCluster.h>
+#include <Storages/SelectQueryInfo.h>
+#include <Storages/StorageSnapshot.h>
+#include <Storages/VirtualColumnUtils.h>
+#include <Storages/removeGroupingFunctionSpecializations.h>
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
+namespace
+{
+
+/// Counts how many times `storage_id` is named as a table in `ast`. The count is taken on the serialized query --
+/// the text a worker actually parses -- not on the query tree, because serialization inlines CTE bodies: a CTE
+/// referenced twice is one node in the tree but two table references in the SQL.
+size_t countTableReferences(const ASTPtr & ast, const StorageID & storage_id)
+{
+    if (!ast)
+        return 0;
+
+    size_t count = 0;
+    if (const auto * identifier = ast->as<ASTTableIdentifier>())
+    {
+        const auto referenced = identifier->getTableId();
+        if (referenced.table_name == storage_id.table_name && referenced.database_name == storage_id.database_name)
+            ++count;
+    }
+
+    for (const auto & child : ast->children)
+        count += countTableReferences(child, storage_id);
+
+    return count;
+}
+
+
+/// Splits a condition into its top-level `and` operands. Anything that is not an `and` is one atom.
+void collectConjunctionAtoms(const QueryTreeNodePtr & node, QueryTreeNodes & atoms)
+{
+    if (const auto * function_node = node->as<FunctionNode>(); function_node && function_node->getFunctionName() == "and")
+    {
+        for (const auto & argument : function_node->getArguments().getNodes())
+            collectConjunctionAtoms(argument, atoms);
+        return;
+    }
+
+    atoms.push_back(node);
+}
+
+/// True when every column this condition reads comes from `driver` and nothing in it has to be executed to be
+/// understood. A subquery is rejected outright: this condition is evaluated while listing the driver's files,
+/// long before there is a pipeline to run one in.
+bool readsOnlyDriverColumns(const QueryTreeNodePtr & node, const TableNode * driver)
+{
+    if (node->as<QueryNode>() || node->as<UnionNode>())
+        return false;
+
+    if (const auto * column_node = node->as<ColumnNode>())
+        return column_node->getColumnSource().get() == driver;
+
+    for (const auto & child : node->getChildren())
+        if (child && !readsOnlyDriverColumns(child, driver))
+            return false;
+
+    return true;
+}
+
+/// The predicate over the driver's own columns, as an ActionsDAG the file listing can prune with.
+///
+/// Correctness rests on two restrictions `findDistributedObjectStorageCandidate` already enforces, and would
+/// break if either were relaxed: 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 QueryNode crossed to reach it is
+/// partition-preserving (`isSafeIntermediateSubquery` -- no GROUP BY, DISTINCT, LIMIT, window), so dropping a
+/// row cannot change what the surviving rows compute.
+///
+/// Returns nothing when no atom qualifies, which simply means every file is listed.
+std::optional<ActionsDAG> buildDriverOnlyFilter(
+    const DistributedObjectStorageCandidate & candidate, const PlannerContextPtr & planner_context)
+{
+    QueryTreeNodes atoms;
+    for (const auto * query_node : candidate.query_nodes_on_driver_path)
+    {
+        if (query_node->hasPrewhere())
+            collectConjunctionAtoms(query_node->getPrewhere(), atoms);
+        if (query_node->hasWhere())
+            collectConjunctionAtoms(query_node->getWhere(), atoms);
+    }
+
+    QueryTreeNodes driver_atoms;
+    for (const auto & atom : atoms)
+        if (readsOnlyDriverColumns(atom, candidate.driver))
+            driver_atoms.push_back(atom->clone());
+
+    if (driver_atoms.empty())
+        return {};
+
+    const auto context = planner_context->getQueryContext();
+    /// mergeConditionNodes always builds an `and`, which needs at least two arguments.
+    auto condition = driver_atoms.size() == 1 ? driver_atoms.front() : mergeConditionNodes(driver_atoms, context);
+
+    /// Passed explicitly so buildFilterInfo does not go looking for this table expression in the planner
+    /// context: the dispatch boundary is planned as one unit and never registers the driver on its own.
+    const auto driver_columns = candidate.driver->getStorageSnapshot()->metadata->getColumns().getNamesOfPhysical();
+    NameSet required_names(driver_columns.begin(), driver_columns.end());
+
+    auto mutable_planner_context = planner_context;
+    auto filter_info = buildFilterInfo(
+        std::move(condition), candidate.driver_table_expression, mutable_planner_context, std::move(required_names));
+
+    return std::move(filter_info.actions);
+}
+
+}
+
+/// This mirrors buildQueryPlanForParallelReplicas (Planner/findParallelReplicasQuery.cpp) step for step:
+/// header of the query -> serialize to SQL -> remote read -> convert the remote header back by position.
+/// Keep the two in sync.
+///
+/// Unlike parallel replicas, the query sent is the one the user wrote: no table expression is rewritten. Which
+/// table drives the dispatch travels beside the query, in `object_storage_distributed_driver_database`/`_table`,
+/// and a worker reads exactly that one table from the initiator's file-task queue.
+std::optional<JoinTreeQueryPlan> buildDistributedObjectStorageQueryPlan(
+    const QueryTreeNodePtr & dispatch_boundary_node,
+    const DistributedObjectStorageCandidate & candidate,
+    const SelectQueryInfo & select_query_info,
+    const PlannerContextPtr & planner_context)
+{
+    const auto context = planner_context->getQueryContext();
+    constexpr auto processed_stage = QueryProcessingStage::WithMergeableState;
+
+    auto * driver_storage = candidate.driver_storage;
+    const auto & driver_storage_snapshot = candidate.driver->getStorageSnapshot();
+    const auto driver_storage_id = candidate.driver->getStorageID();
+
+    /// The header the query produces at this stage, on the initiator and on every worker alike -- the same tree
+    /// serves both, so nothing has to be reconciled by name afterwards.
+    auto [remote_header, new_planner_context] = InterpreterSelectQueryAnalyzer::getSampleBlockAndPlannerContext(
+        dispatch_boundary_node->clone(), context, SelectQueryOptions(processed_stage).analyze());
+
+    /// Strip grouping-function specializations in a clone: the workers re-resolve the generic function
+    /// themselves, but the tree the header came from must keep them.
+    auto query_tree_for_ast = dispatch_boundary_node->clone();
+    removeGroupingFunctionSpecializations(query_tree_for_ast);
+    ASTPtr query_to_send = queryNodeToDistributedSelectQuery(query_tree_for_ast);
+
+    if (!query_to_send->as<ASTSelectQuery>())
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Distributed object-storage dispatch: expected a plain SELECT at the dispatch boundary");
+
+    /// The driver is named by database and table, so it must be unambiguous in the query a worker receives. A
+    /// self-join, or a CTE over the driver referenced more than once, would leave a worker unable to tell which
+    /// occurrence owns the file-task queue -- and reading the queue twice is silently wrong, not an error.
+    /// Fall back to ordinary planning instead.
+    if (countTableReferences(query_to_send, driver_storage_id) != 1)
+        return {};
+
+    /// Travels to the workers with the query. `ReadFromCluster::updateSettings` copies from this context.
+    auto dispatch_context = Context::createCopy(context);
+    dispatch_context->setSetting("object_storage_distributed_driver_database", driver_storage_id.getDatabaseName());
+    dispatch_context->setSetting("object_storage_distributed_driver_table", driver_storage_id.getTableName());
+
+    SelectQueryInfo query_info = select_query_info;
+    query_info.query = query_to_send;
+    query_info.query_tree = dispatch_boundary_node;
+    query_info.planner_context = new_planner_context;
+
+    JoinTreeQueryPlan result;
+    result.stage = processed_stage;
+
+    /// Prunes the driver's file listing. Without it every file of the driver is handed out and each worker
+    /// opens the ones its partitions cannot match only to discard them.
+    std::shared_ptr<const ActionsDAG> driver_filter;
+    if (auto filter_dag = buildDriverOnlyFilter(candidate, planner_context))
+    {
+        VirtualColumnUtils::buildSetsForDAGExcludingGlobalIn(*filter_dag, context);
+        driver_filter = std::make_shared<const ActionsDAG>(std::move(*filter_dag));
+    }
+
+    driver_storage->readPreparedClusterQuery(
+        result.query_plan,
+        driver_storage_snapshot,
+        query_info,
+        dispatch_context,
+        processed_stage,
+        query_to_send,
+        remote_header,
+        std::move(driver_filter));
+
+    /// No converting step, unlike buildQueryPlanForParallelReplicas: that one rewrites the table expression
+    /// before serializing, so its two headers are built from different trees and can diverge. Here the query
+    /// is sent as written and `readPreparedClusterQuery` is given `remote_header` as the source step's own
+    /// output header, so the plan's current header is that same block.
+
+    return result;
+}
+
+}

src/Planner/buildDistributedObjectStorageQueryPlan.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <Planner/findDistributedObjectStorageCandidate.h>
+#include <Planner/PlannerJoinTree.h>
+
+namespace DB
+{
+
+class PlannerContext;
+using PlannerContextPtr = std::shared_ptr<PlannerContext>;
+struct SelectQueryInfo;
+
+/// Builds the whole-query dispatch plan for `candidate`: serializes the query unchanged, announces which table
+/// drives it, and reads the result back through a single ReadFromCluster step at WithMergeableState, so the
+/// caller's normal finalization (MergingAggregated and the rest) applies on top.
+///
+/// No table expression is rewritten. The driving table is named to the workers by database and table, through
+/// `object_storage_distributed_driver_database`/`_table`, and a worker reads that one table from the initiator's
+/// file-task queue while reading every other table in full, locally.
+///
+/// Returns nullopt when the driver cannot be named unambiguously in the serialized query -- it must appear
+/// exactly once. The caller then falls back to ordinary planning.
+///
+/// Structurally the same as buildQueryPlanForParallelReplicas in Planner/findParallelReplicasQuery.cpp,
+/// including the position-based conversion back to the query's header.
+std::optional<JoinTreeQueryPlan> buildDistributedObjectStorageQueryPlan(
+    const QueryTreeNodePtr & dispatch_boundary_node,
+    const DistributedObjectStorageCandidate & candidate,
+    const SelectQueryInfo & select_query_info,
+    const PlannerContextPtr & planner_context);
+
+}