Remove object_storage_cluster_join_mode=local

docs/en/sql-reference/distribution-on-cluster.md

@@ -26,6 +26,6 @@ Receiving new tasks can be resumed with the command `SYSTEM START SWARM MODE`.
 
 ## Joins {#joins}
 
-[`object_storage_cluster_join_mode`](/operations/settings/settings#object_storage_cluster_join_mode) chooses where a `JOIN` of a `*Cluster` table function, or of a table with `object_storage_cluster` set, is executed, and which table is sent to the nodes. With `allow` and `local` the nodes read the left table and the join runs on the initiator. With `global`, an `INNER`, `LEFT`, or `CROSS` join runs on the nodes and the initiator sends the right table (every table after the first, for `CROSS JOIN`). A `RIGHT` or `FULL` join stays on the initiator.
+[`object_storage_cluster_join_mode`](/operations/settings/settings#object_storage_cluster_join_mode) chooses where a `JOIN` of a `*Cluster` table function, or of a table with `object_storage_cluster` set, is executed, and which table is sent to the nodes. With `allow` and `local` the nodes read the left table and the join runs on the initiator. With `global`, an `INNER`, `LEFT`, or `CROSS` join runs on the nodes and the initiator sends the right table (every table after the first, for `CROSS JOIN`). A `RIGHT`, `FULL`, or `PASTE` join stays on the initiator.
 
 The same rules for a `Distributed` table, including which side of a `GLOBAL` `RIGHT` join is sent, are described in [Distributed JOIN](/sql-reference/statements/select/join.md#distributed-join).

src/Core/Settings.cpp

@@ -2131,11 +2131,16 @@ Possible values:
     DECLARE(ObjectStorageClusterJoinMode, object_storage_cluster_join_mode, ObjectStorageClusterJoinMode::ALLOW, R"(
 Where a [JOIN](../../sql-reference/statements/select/join.md#distributed-join) runs when its left table is a cluster table function (`s3Cluster`, `icebergCluster`, and the other `*Cluster` functions) or a table engine with [`object_storage_cluster`](#object_storage_cluster) set.
 
+Restrictions:
+
+- Only applied for `JOIN` and for `IN`/`WHERE` subqueries that reference other tables.
+- Only if the FROM section uses an object storage cluster function or table.
+
 Possible values:
 
-- `allow` — Default. Nodes read the left table and return its rows. The join runs on the initiator. The right table is read there and is not sent to the nodes.
-- `local` — The join runs on the initiator. Nodes read the left table and return its rows. The right table is read on the initiator and is not sent to the nodes.
-- `global` — `INNER`, `LEFT`, and `CROSS` run on the nodes. The initiator reads the right table once and sends it to each node. For `CROSS JOIN`, every table after the first is sent. `RIGHT` and `FULL` stay on the initiator, so unmatched rows of the preserved side are returned once.
+- `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN`. Right table executes first and is added to the secondary query as a temporary table.
+- `allow` — Default value. Reads the left object-storage table on cluster nodes and executes `JOIN` / local `IN` on the initiator, so the right table does not need to exist on remote nodes.
+- `local` — Deprecated legacy alias of `allow`.
 
 See [Distributed JOIN](../../sql-reference/statements/select/join.md#object-storage-cluster-join-mode).
 )", 0) \

src/Core/SettingsEnums.h

@@ -168,9 +168,9 @@ DECLARE_SETTING_ENUM(DistributedProductMode)
 /// The setting for executing object storage cluster function or table JOIN sections.
 enum class ObjectStorageClusterJoinMode : uint8_t
 {
-    LOCAL,       /// Convert to local query
+    LOCAL,       /// Legacy alias of ALLOW: initiator-local join
     GLOBAL,      /// Convert to global query
-    ALLOW        /// Enable
+    ALLOW        /// Initiator-local join when the right table is not on remote nodes
 };
 
 DECLARE_SETTING_ENUM(ObjectStorageClusterJoinMode)

src/Interpreters/InterpreterSelectQuery.cpp

@@ -216,6 +216,8 @@ namespace Setting
     extern const SettingsBool enable_lazy_columns_replication;
     extern const SettingsBool serialize_string_in_memory_with_zero_byte;
     extern const SettingsBool use_hive_partitioning;
+    extern const SettingsBool allow_experimental_analyzer;
+    extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode;
 }
 
 namespace ServerSetting
@@ -788,6 +790,13 @@ InterpreterSelectQuery::InterpreterSelectQuery(
 
     joined_tables.rewriteDistributedInAndJoins(query_ptr);
 
+    if (storage && !settings[Setting::allow_experimental_analyzer]
+        && settings[Setting::object_storage_cluster_join_mode] == ObjectStorageClusterJoinMode::GLOBAL
+        && dynamic_cast<const IStorageCluster *>(storage.get()))
+    {
+        IStorageCluster::rewriteASTForGlobalJoin(query_ptr);
+    }
+
     max_streams = getMaxThreadsForAvailableMemory(
         settings[Setting::max_threads], settings[Setting::max_threads_min_free_memory_per_thread]);
     ASTSelectQuery & query = getSelectQuery();

src/Planner/PlannerJoinTree.cpp

@@ -2289,6 +2289,7 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node,
     size_t joins_count = 0;
     bool is_full_join = false;
     bool is_right_join = false;
+    bool is_paste_join = false;
     bool is_global_join = false;
     bool is_right_join_with_remote_table = false;
     int first_join_pos = -1;
@@ -2320,6 +2321,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node,
                 is_full_join = true;
             if (join_kind == JoinKind::Right)
                 is_right_join = true;
+            if (join_kind == JoinKind::Paste)
+                is_paste_join = true;
 
             if (join_node.getLocality() == JoinLocality::Global)
                 is_global_join = true;
@@ -2408,7 +2411,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node,
       * With `object_storage_cluster_join_mode = 'global'`, `IStorageCluster` must receive a
       * supported full query so it can materialize the right side and send a `GLOBAL JOIN` to
       * remote nodes. `RIGHT JOIN` and `FULL JOIN` stay on the initiator because broadcasting their
-      * preserved right side would make every shard emit the same unmatched rows.
+      * preserved right side would make every shard emit the same unmatched rows. `PASTE JOIN` stays
+      * there too: each shard would pair its own rows with the broadcast right side.
       *
       * `StorageDistributed` has sophisticated query planning logic to handle JOINs and should
       * NOT be wrapped (wrapping breaks tests like 03577_server_constant_folding), except for the
@@ -2421,7 +2425,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node,
             == ObjectStorageClusterJoinMode::GLOBAL
         && joins_count > 0
         && !is_right_join
-        && !is_full_join;
+        && !is_full_join
+        && !is_paste_join;
 
     if (has_multiple_tables && !use_global_join)
     {