Merge 72325abc8e into 44b4bd38b9

docs/en/interfaces/grpc.md

@@ -14,6 +14,7 @@ ClickHouse supports [gRPC](https://grpc.io/) interface. It is an open source rem
 - authentication;
 - sessions;
 - compression;
+- query parameters;
 - parallel queries through the same channel;
 - cancellation of queries;
 - getting progress and logs;
@@ -21,6 +22,14 @@ ClickHouse supports [gRPC](https://grpc.io/) interface. It is an open source rem
 
 The specification of the interface is described in [clickhouse_grpc.proto](https://github.com/ClickHouse/ClickHouse/blob/master/src/Server/grpc_protos/clickhouse_grpc.proto).
 
+Query parameters can be set in the Metadata fields of the Client Context, with `param_` prefix for the key, for example:
+```py
+[
+    ("param_table", "my_table"),
+    ("param_column", "my_column"),
+]
+```
+
 ## gRPC Configuration {#grpc-interface-configuration}
 
 To use the gRPC interface set `grpc_port` in the main [server configuration](../operations/configuration-files.md). Other configuration options see in the following example:
@@ -72,6 +81,7 @@ The client supports the following arguments:
 - `--query QUERY, -q QUERY` – A query to process when using non-interactive mode.
 - `--database DATABASE, -d DATABASE` – A default database. If not specified, the current database set in the server settings is used (`default` by default).
 - `--format OUTPUT_FORMAT, -f OUTPUT_FORMAT` – A result output [format](formats.md). Default value for interactive mode: `PrettyCompact`.
+- `--param PARAM` – Query parameters in `key=value` format. Can be used multiple times.
 - `--debug` – Enables showing debug information.
 
 To run the client in an interactive mode call it without `--query` argument.

src/Server/GRPCServer.cpp

@@ -369,6 +369,11 @@ namespace
             return std::nullopt;
         }
 
+        const std::multimap<::grpc::string_ref, ::grpc::string_ref> & getClientHeaders() const
+        {
+            return grpc_context.client_metadata();
+        }
+
         void setTransportCompression(const TransportCompression & transport_compression)
         {
             grpc_context.set_compression_algorithm(transport_compression.algorithm);
@@ -872,6 +877,16 @@ namespace
 
         query_context = session->makeQueryContext(std::move(client_info));
 
+        /// Extract query params from gRPC client context.
+        for (const auto & [key, value] : responder->getClientHeaders())
+        {
+            if (key.starts_with("param_"))
+            {
+                query_context->setQueryParameter(
+                    String{key.data() + 6, key.size() - 6}, String{value.data(), value.size()});
+            }
+        }
+
         /// Prepare settings.
         SettingsChanges settings_changes;
         for (const auto & [key, value] : query_info.settings())

tests/integration/test_grpc_protocol/test.py

@@ -72,6 +72,7 @@ def query_common(
     session_id="",
     stream_output=False,
     channel=None,
+    query_params={},
 ):
     if type(input_data) is not list:
         input_data = [input_data]
@@ -110,15 +111,24 @@ def query_common(
                 input_data=input_data_part, next_query_info=bool(input_data)
             )
 
+    def metadata():
+        return [("param_" + param, value) for param, value in query_params.items()]
+
     stream_input = len(input_data) > 1
     if stream_input and stream_output:
-        return list(stub.ExecuteQueryWithStreamIO(send_query_info()))
+        return list(
+            stub.ExecuteQueryWithStreamIO(send_query_info(), metadata=metadata())
+        )
     elif stream_input:
-        return [stub.ExecuteQueryWithStreamInput(send_query_info())]
+        return [
+            stub.ExecuteQueryWithStreamInput(send_query_info(), metadata=metadata())
+        ]
     elif stream_output:
-        return list(stub.ExecuteQueryWithStreamOutput(query_info()))
+        return list(
+            stub.ExecuteQueryWithStreamOutput(query_info(), metadata=metadata())
+        )
     else:
-        return [stub.ExecuteQuery(query_info())]
+        return [stub.ExecuteQuery(query_info(), metadata=metadata())]
 
 
 def query_no_errors(*args, **kwargs):
@@ -273,6 +283,24 @@ def test_insert_splitted_row():
     assert query("SELECT a FROM t ORDER BY a") == "1\n2\n3\n4\n5\n6\n"
 
 
+def test_query_params():
+    query(
+        r"CREATE TABLE {table:Identifier} (a UInt8) ENGINE = Memory",
+        query_params={"table": "t"},
+    )
+    query(
+        r"INSERT INTO {table:Identifier} VALUES (1),(2),(3)",
+        query_params={"table": "t"},
+    )
+    assert (
+        query(
+            r"SELECT {column:Identifier} FROM {table:Identifier} ORDER BY {column:Identifier}",
+            query_params={"table": "t", "column": "a"},
+        )
+        == "1\n2\n3\n"
+    )
+
+
 def test_output_format():
     query("CREATE TABLE t (a UInt8) ENGINE = Memory")
     query("INSERT INTO t VALUES (1),(2),(3)")

utils/grpc-client/clickhouse-grpc-client.py

@@ -79,6 +79,7 @@ class ClickHouseGRPCClient(cmd.Cmd):
         settings="",
         verbatim=False,
         show_debug_info=False,
+        params=[],
     ):
         super(ClickHouseGRPCClient, self).__init__(completekey=None)
         self.host = host
@@ -93,6 +94,7 @@ class ClickHouseGRPCClient(cmd.Cmd):
         self.channel = None
         self.stub = None
         self.session_id = None
+        self.params = params
 
     def __enter__(self):
         self.__connect()
@@ -101,8 +103,11 @@ class ClickHouseGRPCClient(cmd.Cmd):
     def __exit__(self, exc_type, exc_value, traceback):
         self.__disconnect()
 
+    def format_metadata(self):
+        return [("param_" + param, value) for param, value in self.params]
+
     # Executes a simple query and returns its output.
-    def get_simple_query_output(self, query_text):
+    def get_simple_query_output(self, query_text, params=[]):
         result = self.stub.ExecuteQuery(
             clickhouse_grpc_pb2.QueryInfo(
                 query=query_text,
@@ -113,7 +118,8 @@ class ClickHouseGRPCClient(cmd.Cmd):
                 settings=self.settings,
                 session_id=self.session_id,
                 query_id=str(uuid.uuid4()),
-            )
+            ),
+            metadata=self.format_metadata(),
         )
         if self.show_debug_info:
             print("\nresult={}".format(result))
@@ -171,7 +177,9 @@ class ClickHouseGRPCClient(cmd.Cmd):
                         if cancel_tries > 0:
                             yield clickhouse_grpc_pb2.QueryInfo(cancel=True)
 
-                for result in self.stub.ExecuteQueryWithStreamIO(send_query_info()):
+                for result in self.stub.ExecuteQueryWithStreamIO(
+                    send_query_info(), metadata=self.format_metadata()
+                ):
                     if self.show_debug_info:
                         print("\nresult={}".format(result))
                     ClickHouseGRPCClient.__check_no_errors(result)
@@ -290,6 +298,11 @@ class ClickHouseGRPCClient(cmd.Cmd):
             readline.write_history_file(histfile)
 
 
+def parse_param(arg):
+    key, value = arg.split("=")
+    return key, value
+
+
 # MAIN
 
 
@@ -341,6 +354,13 @@ def main(args):
         help="Use the specified default format to output the result.",
         default="",
     )
+    parser.add_argument(
+        "--param",
+        action="append",
+        type=parse_param,
+        help="Query parameters in key=value format. Can be used multiple times.",
+        default=[],
+    )
     parser.add_argument(
         "--debug",
         dest="show_debug_info",
@@ -370,6 +390,7 @@ def main(args):
             output_format=output_format,
             verbatim=verbatim,
             show_debug_info=args.show_debug_info,
+            params=args.param,
         ) as client:
             if interactive_mode:
                 client.cmdloop()