forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathTableFunctionRemote.cpp
More file actions
344 lines (259 loc) · 19.7 KB
/
Copy pathTableFunctionRemote.cpp
File metadata and controls
344 lines (259 loc) · 19.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
#include <TableFunctions/TableFunctionRemote.h>
#include <Storages/getStructureOfRemoteTable.h>
#include <Storages/StorageDistributed.h>
#include <Storages/Distributed/DistributedSettings.h>
#include <Interpreters/Cluster.h>
#include <Interpreters/Context.h>
#include <TableFunctions/TableFunctionFactory.h>
#include <TableFunctions/registerTableFunctions.h>
#include <Access/Common/AccessFlags.h>
namespace DB
{
namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
void TableFunctionRemote::parseArguments(const ASTPtr & ast_function, ContextPtr context)
{
ASTs & args_func = ast_function->children;
if (args_func.size() != 1)
throw Exception(help_message, ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
ASTs & args = args_func.at(0)->children;
auto parsed = parseRemoteFunctionArguments(args, context, name, is_cluster_function, secure, help_message);
cluster = std::move(parsed.cluster);
remote_table_id = std::move(parsed.remote_table_id);
sharding_key = std::move(parsed.sharding_key);
remote_table_function_ptr = std::move(parsed.remote_table_function_ptr);
settings_changes = std::move(parsed.settings_changes);
/// Validate the settings early, so that a misspelled setting is reported during parsing
/// rather than at the first read from the storage.
DistributedSettings{}.applyChanges(settings_changes);
}
StoragePtr TableFunctionRemote::executeImpl(const ASTPtr & /*ast_function*/, ContextPtr context, const std::string & table_name, ColumnsDescription cached_columns, bool is_insert_query) const
{
/// StorageDistributed supports mismatching structure of remote table, so we can use outdated structure for CREATE ... AS remote(...)
/// without additional conversion in StorageTableFunctionProxy
if (cached_columns.empty())
cached_columns = getActualTableStructure(context, is_insert_query);
chassert(cluster);
bool has_local_shard = false;
for (const auto & shard_info : cluster->getShardsInfo())
{
if (shard_info.isLocal())
{
has_local_shard = true;
break;
}
}
if (has_local_shard && !is_insert_query)
context->checkAccess(AccessType::SELECT, remote_table_id);
else if (has_local_shard)
context->checkAccess(AccessType::INSERT, remote_table_id);
DistributedSettings distributed_settings;
distributed_settings.applyChanges(settings_changes);
StoragePtr res = std::make_shared<StorageDistributed>(
StorageID(getDatabaseName(), table_name),
cached_columns,
ConstraintsDescription{},
String{},
remote_table_id.database_name,
remote_table_id.table_name,
String{},
context,
sharding_key,
String{},
String{},
distributed_settings,
LoadingStrictnessLevel::CREATE,
cluster,
remote_table_function_ptr,
!is_cluster_function);
res->startup();
return res;
}
ColumnsDescription TableFunctionRemote::getActualTableStructure(ContextPtr context, bool /*is_insert_query*/) const
{
chassert(cluster);
return getStructureOfRemoteTable(*cluster, remote_table_id, context, remote_table_function_ptr);
}
TableFunctionRemote::TableFunctionRemote(const std::string & name_, bool secure_)
: name{name_}, secure{secure_}
{
is_cluster_function = (name == "cluster" || name == "clusterAllReplicas");
help_message = PreformattedMessage::create(
"Table function '{}' requires from {} to {} parameters: "
"{}",
name,
is_cluster_function ? 0 : 1,
is_cluster_function ? 4 : 6,
is_cluster_function ? "[<cluster name or default if not specified>, [<database.table> | [<name of remote database>, <name of remote table>] | <table function>]] [, sharding_key]"
: "<addresses pattern> [, <name of remote database>, <name of remote table>] [, username[, password], sharding_key]");
}
void registerTableFunctionRemote(TableFunctionFactory & factory)
{
factory.registerFunction("remote", {[] () -> TableFunctionPtr { return std::make_shared<TableFunctionRemote>("remote"); }, {.description = R"DOCS_MD(
Table function `remote` allows to access remote servers on-the-fly, i.e. without creating a [Distributed](/reference/engines/table-engines/special/distributed) table. Table function `remoteSecure` is same as `remote` but over a secure connection.
Both functions can be used in `SELECT` and `INSERT` queries when the target is an ordinary `db`/`table`. When the target is itself a table function (for example `remote('127.0.0.1', numbers(10))`), the table is read-only: there is no remote table to insert into, so `INSERT` is rejected with a `NOT_IMPLEMENTED` exception.
## Syntax {#syntax}
```sql
remote(addresses_expr, [db, table, user [, password], sharding_key][, SETTINGS name = value, ...])
remote(addresses_expr, [db.table, user [, password], sharding_key][, SETTINGS name = value, ...])
remote(named_collection[, option=value [,..]][, SETTINGS name = value, ...])
remoteSecure(addresses_expr, [db, table, user [, password], sharding_key][, SETTINGS name = value, ...])
remoteSecure(addresses_expr, [db.table, user [, password], sharding_key][, SETTINGS name = value, ...])
remoteSecure(named_collection[, option=value [,..]][, SETTINGS name = value, ...])
```
## Parameters {#parameters}
| Argument | Description |
|----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `addresses_expr` | A remote server address or an expression that generates multiple addresses of remote servers. Format: `host` or `host:port`.<br/><br/> The `host` can be specified as a server name, or as a IPv4 or IPv6 address. An IPv6 address must be specified in `[]`.<br/><br/> The `port` is the TCP port on the remote server. If the port is omitted, it uses [tcp_port](/reference/settings/server-settings/settings/tcp-port#tcp_port) from the server config file for table function `remote` (by default, 9000) and [tcp_port_secure](/reference/settings/server-settings/settings/tcp-port#tcp_port_secure) for table function `remoteSecure` (by default, 9440).<br/><br/> For IPv6 addresses, a port is required.<br/><br/> If only parameter `addresses_expr` is specified, `db` and `table` will use `system.one` by default.<br/><br/> Type: [String](/reference/data-types/string). |
| `db` | Database name. Type: [String](/reference/data-types/string). |
| `table` | Table name. Type: [String](/reference/data-types/string). |
| `user` | User name. If not specified, `default` is used. Type: [String](/reference/data-types/string). |
| `password` | User password. If not specified, an empty password is used. Type: [String](/reference/data-types/string). |
| `sharding_key` | Sharding key to support distributing data across nodes. For example: `insert into remote('127.0.0.1:9000,127.0.0.2', db, table, 'default', rand())`. Type: [UInt32](/reference/data-types/int-uint). |
| `SETTINGS name = value, ...` | Settings of the Distributed table created by the function, for example `skip_unavailable_shards`. Optional. A setting specified in the query has priority over it. This clause is accepted only inside the table function; the `Remote` and `RemoteSecure` table engines take the same settings after the engine definition, see [Remote and RemoteSecure engines](/reference/engines/table-engines/special/distributed#distributed-remote-engines). |
Arguments also can be passed using [named collections](/concepts/features/configuration/server-config/named-collections).
## Returned value {#returned-value}
A table located on a remote server.
## Usage {#usage}
As table functions `remote` and `remoteSecure` re-establish the connection for each request, it is recommended to use a `Distributed` table instead. Also, if hostnames are set, the names are resolved, and errors are not counted when working with various replicas. When processing a large number of queries, always create the `Distributed` table ahead of time, and do not use the `remote` table function.
The `remote` table function can be useful in the following cases:
- One-time data migration from one system to another
- Accessing a specific server for data comparison, debugging, and testing, i.e. ad-hoc connections.
- Queries between various ClickHouse clusters for research purposes.
- Infrequent distributed requests that are made manually.
- Distributed requests where the set of servers is re-defined each time.
The same parameters can be used with the `Remote` and `RemoteSecure` table engines to create a persistent table instead of an ad-hoc one, see [Remote and RemoteSecure engines](/reference/engines/table-engines/special/distributed#distributed-remote-engines). The only difference is the `SETTINGS` clause: the engines take it after the engine definition, `ENGINE = Remote(...) SETTINGS skip_unavailable_shards = 1`, not among the arguments.
### Addresses {#addresses}
```text
example01-01-1
example01-01-1:9440
example01-01-1:9000
localhost
127.0.0.1
[::]:9440
[::]:9000
[2a02:6b8:0:1111::11]:9000
```
Multiple addresses can be comma-separated. In this case, ClickHouse will use distributed processing and send the query to all specified addresses (like shards with different data). Example:
```text
example01-01-1,example01-02-1
```
## Examples {#examples}
### Selecting data from a remote server: {#selecting-data-from-a-remote-server}
```sql
SELECT * FROM remote('127.0.0.1', db.remote_engine_table) LIMIT 3;
```
Or using [named collections](/concepts/features/configuration/server-config/named-collections):
```sql
CREATE NAMED COLLECTION creds AS
host = '127.0.0.1',
database = 'db';
SELECT * FROM remote(creds, table='remote_engine_table') LIMIT 3;
```
### Inserting data into a table on a remote server: {#inserting-data-into-a-table-on-a-remote-server}
```sql
CREATE TABLE remote_table (name String, value UInt32) ENGINE=Memory;
INSERT INTO FUNCTION remote('127.0.0.1', currentDatabase(), 'remote_table') VALUES ('test', 42);
SELECT * FROM remote_table;
```
### Migration of tables from one system to another: {#migration-of-tables-from-one-system-to-another}
This example uses one table from a sample dataset. The database is `imdb`, and the table is `actors`.
#### On the source ClickHouse system (the system that currently hosts the data) {#on-the-source-clickhouse-system-the-system-that-currently-hosts-the-data}
- Verify the source database and table name (`imdb.actors`)
```sql
show databases
```
```sql
show tables in imdb
```
- Get the CREATE TABLE statement from the source:
```sql
SELECT create_table_query
FROM system.tables
WHERE database = 'imdb' AND table = 'actors'
```
Response
```sql
CREATE TABLE imdb.actors (`id` UInt32,
`first_name` String,
`last_name` String,
`gender` FixedString(1))
ENGINE = MergeTree
ORDER BY (id, first_name, last_name, gender);
```
#### On the destination ClickHouse system {#on-the-destination-clickhouse-system}
- Create the destination database:
```sql
CREATE DATABASE imdb
```
- Using the CREATE TABLE statement from the source, create the destination:
```sql
CREATE TABLE imdb.actors (`id` UInt32,
`first_name` String,
`last_name` String,
`gender` FixedString(1))
ENGINE = MergeTree
ORDER BY (id, first_name, last_name, gender);
```
#### Back on the source deployment {#back-on-the-source-deployment}
Insert into the new database and table created on the remote system. You will need the host, port, username, password, destination database, and destination table.
```sql
INSERT INTO FUNCTION
remoteSecure('remote.clickhouse.cloud:9440', 'imdb.actors', 'USER', 'PASSWORD')
SELECT * from imdb.actors
```
## Globbing {#globs-in-addresses}
Patterns in `{ }` are used to generate a set of shards and to specify replicas. If there are multiple pairs of `{ }`, then the direct product of the corresponding sets is generated.
The following pattern types are supported.
- `{a,b,c}` - Represents any of alternative strings `a`, `b` or `c`. The pattern is replaced with `a` in the first shard address and replaced with `b` in the second shard address and so on. For instance, `example0{1,2}-1` generates addresses `example01-1` and `example02-1`.
- `{N..M}` - A range of numbers. This pattern generates shard addresses with incrementing indices from `N` to (and including) `M`. For instance, `example0{1..2}-1` generates `example01-1` and `example02-1`.
- `{0n..0m}` - A range of numbers with leading zeroes. This pattern preserves leading zeroes in indices. For instance, `example{01..03}-1` generates `example01-1`, `example02-1` and `example03-1`.
- `{a|b}` - Any number of variants separated by a `|`. The pattern specifies replicas. For instance, `example01-{1|2}` generates replicas `example01-1` and `example01-2`.
The query will be sent to the first healthy replica. However, for `remote` the replicas are iterated in the order currently set in the [load_balancing](/reference/settings/session-settings/load-balancing#load_balancing) setting.
The number of generated addresses is limited by [table_function_remote_max_addresses](/reference/settings/session-settings/table#table_function_remote_max_addresses) setting.
)DOCS_MD", .category = FunctionDocumentation::Category::TableFunction}});
factory.registerFunction("remoteSecure", {[] () -> TableFunctionPtr { return std::make_shared<TableFunctionRemote>("remote", /* secure = */ true); }, {.description = R"DOC(Like the remote table function, but establishes a TLS-encrypted (secure) connection to the remote server.)DOC", .category = FunctionDocumentation::Category::TableFunction}});
factory.registerFunction("cluster", {[] () -> TableFunctionPtr { return std::make_shared<TableFunctionRemote>("cluster"); }, {.description = R"DOCS_MD(
Allows accessing all shards (configured in the `remote_servers` section) of a cluster without creating a [Distributed](/reference/engines/table-engines/special/distributed) table. Only one replica of each shard is queried.
`clusterAllReplicas` function — same as `cluster`, but all replicas are queried. Each replica in a cluster is used as a separate shard/connection.
<Note>
All available clusters are listed in the [system.clusters](/reference/system-tables/clusters) table.
</Note>
## Syntax {#syntax}
```sql
cluster(['cluster_name', db.table, sharding_key][, SETTINGS name = value, ...])
cluster(['cluster_name', db, table, sharding_key][, SETTINGS name = value, ...])
clusterAllReplicas(['cluster_name', db.table, sharding_key][, SETTINGS name = value, ...])
clusterAllReplicas(['cluster_name', db, table, sharding_key][, SETTINGS name = value, ...])
```
## Arguments {#arguments}
| Arguments | Type |
|-----------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------|
| `cluster_name` | Name of a cluster that is used to build a set of addresses and connection parameters to remote and local servers, set `default` if not specified. |
| `db.table` or `db`, `table` | Name of a database and a table. |
| `sharding_key` | A sharding key. Optional. Needs to be specified if the cluster has more than one shard. |
| `SETTINGS name = value, ...` | Settings of the Distributed table created by the function, for example `skip_unavailable_shards`. Optional. A setting specified in the query has priority over it. |
## Returned value {#returned-value}
The dataset from clusters.
## Using macros {#using-macros}
`cluster_name` can contain macros — substitution in `{}`. The substituted value is taken from the [macros](/reference/settings/server-settings/settings/other#macros) section of the server configuration file.
Example:
```sql
SELECT * FROM cluster('{cluster}', default.example_table);
```
## Usage and recommendations {#usage_recommendations}
Using the `cluster` and `clusterAllReplicas` table functions are less efficient than creating a `Distributed` table because in this case, the server connection is re-established for every request. When processing a large number of queries, please always create the `Distributed` table ahead of time, and do not use the `cluster` and `clusterAllReplicas` table functions.
The `cluster` and `clusterAllReplicas` table functions can be useful in the following cases:
- Accessing a specific cluster for data comparison, debugging, and testing.
- Queries to various ClickHouse clusters and replicas for research purposes.
- Infrequent distributed requests that are made manually.
Connection settings like `host`, `port`, `user`, `password`, `compression`, `secure` are taken from `<remote_servers>` config section. See details in [Distributed engine](/reference/engines/table-engines/special/distributed).
## Related {#related}
- [skip_unavailable_shards](/reference/settings/session-settings/skip-unavailable-shards#skip_unavailable_shards)
- [load_balancing](/reference/settings/session-settings/load-balancing#load_balancing)
)DOCS_MD", .category = FunctionDocumentation::Category::TableFunction}, {.allow_readonly = true}});
factory.registerFunction("clusterAllReplicas", {[] () -> TableFunctionPtr { return std::make_shared<TableFunctionRemote>("clusterAllReplicas"); }, {.description = R"DOC(Like the cluster table function, but queries all replicas of every shard in the cluster instead of a single replica per shard.)DOC", .category = FunctionDocumentation::Category::TableFunction}, {.allow_readonly = true}});
}
}