Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions docs/en/antalya/swarm.md
Original file line number Diff line number Diff line change
Expand Up @@ -281,14 +281,13 @@ the same node while the cluster is stable, which helps local caches. See
### Joins against local tables {#joins}

A swarm query can `JOIN` a table that exists only on the initiator. Workers do not have that table,
so you must choose how the join is rewritten with `object_storage_cluster_join_mode`
(requires `allow_experimental_analyzer = 1` unless the value is `allow`):
so you must choose how the join is rewritten with `object_storage_cluster_join_mode`:

| Value | Behaviour |
| --- | --- |
| `allow` (default) | Leave the join as written. Use this when both sides are object-storage cluster scans, or when workers can see the right-hand table. |
| `local` | Send only the object-storage scan to the swarm; perform the join on the initiator. |
| `global` | Read the right-hand side on the initiator first and ship it to workers as a temporary table. |
| `allow` (default) | Send only the object-storage scan to the swarm; perform the join on the initiator. The right table does not need to exist on workers. |
| `local` | Deprecated alias of `allow`. |
| `global` | Read the right-hand side on the initiator first and ship it to workers as a temporary table. `INNER`, `LEFT`, and `CROSS` run on the workers. `RIGHT` and `FULL` stay on the initiator. |

```sql
SELECT *
Expand Down Expand Up @@ -467,5 +466,5 @@ connections drain.
| `object_storage_max_nodes` | Cap on how many hosts from that cluster run a given query. `0` = all. |
| `object_storage_remote_initiator` | Run the cluster query on a random swarm node instead of coordinating from the client-facing initiator. |
| `object_storage_remote_initiator_cluster` | Cluster used only to pick that remote initiator. When empty, `object_storage_cluster` is used. |
| `object_storage_cluster_join_mode` | How to rewrite `JOIN`s when workers cannot see the right-hand table: `allow`, `local`, `global`. |
| `object_storage_cluster_join_mode` | How to rewrite `JOIN`s when workers cannot see the right-hand table: `allow` (initiator-local join), `local` (deprecated alias of `allow`), `global`. |
| `lock_object_storage_task_distribution_ms` | How long a free worker waits before stealing another node's files. Higher values favor cache locality. Default `500`. |
2 changes: 1 addition & 1 deletion docs/en/sql-reference/distribution-on-cluster.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` the nodes read the left table and the join runs on the initiator. `local` is a deprecated alias of `allow`. 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).
9 changes: 7 additions & 2 deletions src/Core/Settings.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2306,10 +2306,15 @@ 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.
- `allow` — Default. 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`.
- `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.

See [Distributed JOIN](../../sql-reference/statements/select/join.md#object-storage-cluster-join-mode).
Expand Down
4 changes: 2 additions & 2 deletions src/Core/SettingsEnums.h
Original file line number Diff line number Diff line change
Expand Up @@ -171,9 +171,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)
Expand Down
8 changes: 8 additions & 0 deletions src/Interpreters/InterpreterSelectQuery.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,7 @@ 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 SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode;
}

namespace ServerSetting
Expand Down Expand Up @@ -815,6 +816,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();
Expand Down
9 changes: 7 additions & 2 deletions src/Planner/PlannerJoinTree.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -3600,6 +3600,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;
Expand Down Expand Up @@ -3631,6 +3632,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;
Expand Down Expand Up @@ -3719,7 +3722,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
Expand All @@ -3732,7 +3736,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)
{
Expand Down
Loading
Loading