Fix bucket-merge scheduling and ReadFromCluster filter staleness

src/Core/Settings.cpp

@@ -2138,6 +2138,22 @@ 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.
+)", 0) \
+    DECLARE(Bool, object_storage_cluster_bypass_join_wrap, false, R"(
+Experimental. When enabled, a DataLake table backed by a shared catalog (e.g.
+Iceberg, resolved as `StorageObjectStorageCluster`) becomes eligible to be chosen
+as a parallel-replicas driver for a `JOIN` query, the same way a `MergeTree` table
+already can be: the same candidate-selection machinery used for `MergeTree`
+(`findQueryForParallelReplicas`/`findTableForParallelReplicas` for a driver reachable
+through a CTE or subquery, and the query planner's existing leftmost/rightmost
+join-shape eligibility check for a driver that is an immediate JOIN operand) picks
+the driver, and the whole `JOIN` (and any partial aggregation) is executed on the
+remote cluster nodes instead of pulling all probe-side rows back to the initiator.
+
+This is only safe when every table expression in the query is resolvable on all
+remote nodes of the cluster (e.g. tables backed by the same shared catalog). Off by
+default (`object_storage_cluster_bypass_join_wrap=false`), leaving normal plans for
+unrelated queries unaffected.
 )", 0) \
     \
     DECLARE(UInt64, max_concurrent_queries_for_all_users, 0, R"(

src/Core/SettingsChangesHistory.cpp

@@ -42,6 +42,7 @@ 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_cluster_bypass_join_wrap", false, false, "Experimental setting to allow IStorageCluster sources (e.g. Iceberg StorageObjectStorageCluster) to receive the full JOIN query instead of being wrapped in a subquery, so JOIN + partial aggregation can run on parallel-replica workers."},
         });
 
         addSettingsChanges(settings_changes_history, "26.6",

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -751,6 +751,22 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
     if (cluster_name.empty() && can_use_parallel_replicas && !is_secondary_query)
         cluster_name = parallel_replicas_cluster_name;
 
+    /// EXPERIMENTAL diagnostic (see object_storage_cluster_bypass_join_wrap in PlannerJoinTree.cpp): trace
+    /// why can_use_parallel_replicas may end up false for a DataLake table under a JOIN.
+    LOG_WARNING(
+        log,
+        "DLPR table={} cluster_for_pr='{}' parallel_cluster_engines={} "
+        "can_task_pr={} is_distributed={} query_kind={} "
+        "can_use_pr={} final_cluster='{}'",
+        name,
+        parallel_replicas_cluster_name,
+        static_cast<bool>(query_settings[Setting::parallel_replicas_for_cluster_engines]),
+        context_->canUseTaskBasedParallelReplicas(),
+        context_->isDistributed(),
+        static_cast<int>(context_->getClientInfo().query_kind),
+        can_use_parallel_replicas,
+        cluster_name);
+
     auto storage_cluster = std::make_shared<StorageObjectStorageCluster>(
         cluster_name,
         configuration,

src/Planner/Planner.cpp

@@ -1913,11 +1913,23 @@ Planner::Planner(const QueryTreeNodePtr & query_tree_,
     : query_tree(query_tree_)
     , select_query_options(select_query_options_)
     , planner_context(buildPlannerContext(query_tree, select_query_options,
-        std::make_shared<GlobalPlannerContext>(
-            findQueryForParallelReplicas(query_tree, select_query_options),
-            findTableForParallelReplicas(query_tree, select_query_options),
-            findTableUnionForParallelReplicas(query_tree, select_query_options),
-            collectFiltersForAnalysis(query_tree, select_query_options, post_filter_))))
+        [&]
+        {
+            /// Computed once and reused below: GlobalPlannerContext::parallel_replicas_candidate_driver is
+            /// derived directly from this exact candidate (findParallelReplicasCandidateDriver()), not
+            /// independently re-derived from query_tree -- see that function's own comment for why. Must NOT
+            /// use GlobalPlannerContext::parallel_replicas_table for this: that field comes from the public
+            /// findTableForParallelReplicas() overload, whose follower-only gate makes it nullptr
+            /// unconditionally on a normal initiator (it exists for PlannerJoinTree.cpp's View/
+            /// MaterializedView follower-recursion-safety check, an unrelated purpose).
+            const QueryNode * parallel_replicas_node = findQueryForParallelReplicas(query_tree, select_query_options);
+            return std::make_shared<GlobalPlannerContext>(
+                parallel_replicas_node,
+                findTableForParallelReplicas(query_tree, select_query_options),
+                findTableUnionForParallelReplicas(query_tree, select_query_options),
+                collectFiltersForAnalysis(query_tree, select_query_options, post_filter_),
+                findParallelReplicasCandidateDriver(parallel_replicas_node));
+        }()))
 {
 }
 
@@ -2295,7 +2307,20 @@ void Planner::buildPlanForQueryNode()
     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);
+        /// GlobalPlannerContext::parallel_replicas_candidate_driver was found by
+        /// findParallelReplicasCandidateDriver(parallel_replicas_node) -- an initiator-safe lookup derived
+        /// directly from parallel_replicas_node itself (see findQueryForParallelReplicas.h; NOT
+        /// parallel_replicas_table, which is nullptr on a normal initiator by design -- it exists for
+        /// PlannerJoinTree.cpp's unrelated follower-recursion-safety check). Dispatch to the matching
+        /// execution backend based on the driver's storage kind: MergeTree keeps its existing
+        /// task-based-coordinator path; an object-storage-cluster driver goes through the same
+        /// exact-QueryTree-replacement/ReadFromCluster backend PlannerJoinTree.cpp's leftmost-driver case uses
+        /// below, so ObjectStorage execution is identical regardless of where the driver sits in the tree.
+        const TableNode * driver_table = planner_context->getGlobalPlannerContext()->parallel_replicas_candidate_driver;
+        if (driver_table && dynamic_cast<const StorageObjectStorageCluster *>(driver_table->getStorage().get()))
+            join_tree_query_plan = buildQueryPlanForObjectStorageCluster(query_tree, *driver_table, select_query_info, planner_context);
+        else
+            join_tree_query_plan = buildQueryPlanForParallelReplicas(query_node, planner_context, select_query_info.storage_limits);
     }
     else
     {

src/Planner/PlannerContext.h

@@ -41,10 +41,12 @@ class GlobalPlannerContext
         const QueryNode * parallel_replicas_node_,
         const TableNode * parallel_replicas_table_,
         const UnionNode * parallel_replicas_table_union_,
-        FiltersForTableExpressionMap filters_for_table_expressions_)
+        FiltersForTableExpressionMap filters_for_table_expressions_,
+        const TableNode * parallel_replicas_candidate_driver_ = nullptr)
         : parallel_replicas_node(parallel_replicas_node_)
         , parallel_replicas_table(parallel_replicas_table_)
         , parallel_replicas_table_union(parallel_replicas_table_union_)
+        , parallel_replicas_candidate_driver(parallel_replicas_candidate_driver_)
         , filters_for_table_expressions(std::move(filters_for_table_expressions_))
     {
     }
@@ -84,6 +86,14 @@ class GlobalPlannerContext
     /// UNION node whose every child query reads from a table eligible for parallel replicas.
     /// When set, each branch retains parallel replicas reading instead of having it disabled.
     const UnionNode * const parallel_replicas_table_union = nullptr;
+    /// Driver TableNode associated with parallel_replicas_node, found via findParallelReplicasCandidateDriver()
+    /// (findQueryForParallelReplicas.h) -- an initiator-safe lookup derived directly from parallel_replicas_node
+    /// itself. Deliberately separate from parallel_replicas_table above: that field comes from the public,
+    /// follower-gated findTableForParallelReplicas() overload and is nullptr on a normal initiator by design
+    /// (it exists for PlannerJoinTree.cpp's unrelated View/MaterializedView follower-recursion-safety check).
+    /// Used by Planner::buildPlanForQueryNode() to pick the execution backend (MergeTree vs.
+    /// object-storage-cluster) for whatever table drives parallel_replicas_node's dispatch.
+    const TableNode * const parallel_replicas_candidate_driver = nullptr;
 
     const FiltersForTableExpressionMap filters_for_table_expressions;