GLOBAL IN, and
GLOBAL NOT IN operators are covered separately, since their functionality is quite rich.
The left side of the operator is either a single column or a tuple.
SELECT UserID IN (123, 456) FROM ... SELECT (CounterID, UserID) IN ((34, 123), (101500, 456)) FROM ...
If the left side is a single column that is in the index, and the right side is a set of constants, the system uses the index for processing the query.
Don’t list too many values explicitly (i.e. millions). If a data set is large, put it in a temporary table (for example, see the section “External data for query processing”), then use a subquery.
The right side of the operator can be a set of constant expressions, a set of tuples with constant expressions (shown in the examples above), or the name of a database table or SELECT subquery in brackets.
If the right side of the operator is the name of a table (for example,
UserID IN users), this is equivalent to the subquery
UserID IN (SELECT * FROM users). Use this when working with external data that is sent along with the query. For example, the query can be sent together with a set of user IDs loaded to the ‘users’ temporary table, which should be filtered.
If the right side of the operator is a table name that has the Set engine (a prepared data set that is always in RAM), the data set will not be created over again for each query.
The subquery may specify more than one column for filtering tuples.
SELECT (CounterID, UserID) IN (SELECT CounterID, UserID FROM ...) FROM ...
The columns to the left and right of the IN operator should have the same type.
The IN operator and subquery may occur in any part of the query, including in aggregate functions and lambda functions.
SELECT EventDate, avg(UserID IN ( SELECT UserID FROM test.hits WHERE EventDate = toDate('2014-03-17') )) AS ratio FROM test.hits GROUP BY EventDate ORDER BY EventDate ASC
┌──EventDate─┬────ratio─┐ │ 2014-03-17 │ 1 │ │ 2014-03-18 │ 0.807696 │ │ 2014-03-19 │ 0.755406 │ │ 2014-03-20 │ 0.723218 │ │ 2014-03-21 │ 0.697021 │ │ 2014-03-22 │ 0.647851 │ │ 2014-03-23 │ 0.648416 │ └────────────┴──────────┘
For each day after March 17th, count the percentage of pageviews made by users who visited the site on March 17th.
A subquery in the IN clause is always run just one time on a single server. There are no dependent subqueries.
During request processing, the
IN operator assumes that the result of an operation with NULL always equals
0, regardless of whether
NULL is on the right or left side of the operator.
NULL values are not included in any dataset, do not correspond to each other and cannot be compared if transform_null_in = 0.
Here is an example with the
┌─x─┬────y─┐ │ 1 │ ᴺᵁᴸᴸ │ │ 2 │ 3 │ └───┴──────┘
Running the query
SELECT x FROM t_null WHERE y IN (NULL,3) gives you the following result:
┌─x─┐ │ 2 │ └───┘
You can see that the row in which
y = NULL is thrown out of the query results. This is because ClickHouse can’t decide whether
NULL is included in the
(NULL,3) set, returns
0 as the result of the operation, and
SELECT excludes this row from the final output.
SELECT y IN (NULL, 3) FROM t_null
┌─in(y, tuple(NULL, 3))─┐ │ 0 │ │ 1 │ └───────────────────────┘
There are two options for IN-s with subqueries (similar to JOINs): normal
GLOBAL IN /
GLOBAL JOIN. They differ in how they are run for distributed query processing.
Remember that the algorithms described below may work differently depending on the settings
When using the regular IN, the query is sent to remote servers, and each of them runs the subqueries in the
GLOBAL IN /
GLOBAL JOINs, first all the subqueries are run for
GLOBAL IN /
GLOBAL JOINs, and the results are collected in temporary tables. Then the temporary tables are sent to each remote server, where the queries are run using this temporary data.
For a non-distributed query, use the regular
Be careful when using subqueries in the
JOIN clauses for distributed query processing.
Let’s look at some examples. Assume that each server in the cluster has a normal local_table. Each server also has a distributed_table table with the Distributed type, which looks at all the servers in the cluster.
For a query to the distributed_table, the query will be sent to all the remote servers and run on them using the local_table.
For example, the query
SELECT uniq(UserID) FROM distributed_table
will be sent to all remote servers as
SELECT uniq(UserID) FROM local_table
and run on each of them in parallel, until it reaches the stage where intermediate results can be combined. Then the intermediate results will be returned to the requestor server and merged on it, and the final result will be sent to the client.
Now let’s examine a query with IN:
SELECT uniq(UserID) FROM distributed_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM local_table WHERE CounterID = 34)
- Calculation of the intersection of audiences of two sites.
This query will be sent to all remote servers as
SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM local_table WHERE CounterID = 34)
In other words, the data set in the IN clause will be collected on each server independently, only across the data that is stored locally on each of the servers.
This will work correctly and optimally if you are prepared for this case and have spread data across the cluster servers such that the data for a single UserID resides entirely on a single server. In this case, all the necessary data will be available locally on each server. Otherwise, the result will be inaccurate. We refer to this variation of the query as “local IN”.
To correct how the query works when data is spread randomly across the cluster servers, you could specify distributed_table inside a subquery. The query would look like this:
SELECT uniq(UserID) FROM distributed_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)
This query will be sent to all remote servers as
SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)
The subquery will begin running on each remote server. Since the subquery uses a distributed table, the subquery that is on each remote server will be resent to every remote server as
SELECT UserID FROM local_table WHERE CounterID = 34
For example, if you have a cluster of 100 servers, executing the entire query will require 10,000 elementary requests, which is generally considered unacceptable.
In such cases, you should always use GLOBAL IN instead of IN. Let’s look at how it works for the query
SELECT uniq(UserID) FROM distributed_table WHERE CounterID = 101500 AND UserID GLOBAL IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)
The requestor server will run the subquery
SELECT UserID FROM distributed_table WHERE CounterID = 34
and the result will be put in a temporary table in RAM. Then the request will be sent to each remote server as
SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID GLOBAL IN _data1
and the temporary table
_data1 will be sent to every remote server with the query (the name of the temporary table is implementation-defined).
This is more optimal than using the normal IN. However, keep the following points in mind:
- When creating a temporary table, data is not made unique. To reduce the volume of data transmitted over the network, specify DISTINCT in the subquery. (You don’t need to do this for a normal IN.)
- The temporary table will be sent to all the remote servers. Transmission does not account for network topology. For example, if 10 remote servers reside in a datacenter that is very remote in relation to the requestor server, the data will be sent 10 times over the channel to the remote datacenter. Try to avoid large data sets when using GLOBAL IN.
- When transmitting data to remote servers, restrictions on network bandwidth are not configurable. You might overload the network.
- Try to distribute data across servers so that you don’t need to use GLOBAL IN on a regular basis.
- If you need to use GLOBAL IN often, plan the location of the ClickHouse cluster so that a single group of replicas resides in no more than one data center with a fast network between them, so that a query can be processed entirely within a single data center.
It also makes sense to specify a local table in the
GLOBAL IN clause, in case this local table is only available on the requestor server and you want to use data from it on remote servers.
When max_parallel_replicas is greater than 1, distributed queries are further transformed. For example, the following:
SEELECT CounterID, count() FROM distributed_table_1 WHERE UserID IN (SELECT UserID FROM local_table_2 WHERE CounterID < 100) SETTINGS max_parallel_replicas=3
is transformed on each server into
SELECT CounterID, count() FROM local_table_1 WHERE UserID IN (SELECT UserID FROM local_table_2 WHERE CounterID < 100) SETTINGS parallel_replicas_count=3, parallel_replicas_offset=M
where M is between 1 and 3 depending on which replica the local query is executing on. These settings affect every MergeTree-family table in the query and have the same effect as applying
SAMPLE 1/3 OFFSET (M-1)/3 on each table.
Therefore adding the max_parallel_replicas setting will only produce correct results if both tables have the same replication scheme and are sampled by UserID or a subkey of it. In particular, if local_table_2 does not have a sampling key, incorrect results will be produced. The same rule applies to JOIN.
One workaround if local_table_2 doesn't meet the requirements, is to use
GLOBAL IN or