Merge pull request ClickHouse#80073 from nibblerenush/log_udf_name

Add executable  UDF names to the `used_functions` column in the `system.query_log`.

Author: alexey-milovidov

src/Functions/UserDefined/UserDefinedExecutableFunctionFactory.cpp

@@ -2,9 +2,11 @@
 
 #include <filesystem>
 
+#include <Common/CurrentThread.h>
 #include <Common/filesystemHelpers.h>
 #include <Common/FieldVisitorToString.h>
 #include <Common/quoteString.h>
+#include <Core/Settings.h>
 #include <DataTypes/FieldToDataType.h>
 
 #include <Processors/Sources/ShellCommandSource.h>
@@ -28,6 +30,11 @@ namespace ErrorCodes
     extern const int BAD_ARGUMENTS;
 }
 
+namespace Setting
+{
+    extern const SettingsBool log_queries;
+}
+
 namespace
 {
 
@@ -245,6 +252,14 @@ FunctionOverloadResolverPtr UserDefinedExecutableFunctionFactory::get(const Stri
     const auto & loader = context->getExternalUserDefinedExecutableFunctionsLoader();
     auto executable_function = std::static_pointer_cast<const UserDefinedExecutableFunction>(loader.load(function_name));
     auto function = std::make_shared<UserDefinedFunction>(std::move(executable_function), std::move(context), std::move(parameters));
+
+    if (CurrentThread::isInitialized())
+    {
+        auto query_context = CurrentThread::get().getQueryContext();
+        if (query_context && query_context->getSettingsRef()[Setting::log_queries])
+            query_context->addQueryFactoriesInfo(Context::QueryLogFactories::Function, function_name);
+    }
+
     return std::make_unique<FunctionToOverloadResolverAdaptor>(std::move(function));
 }
 
@@ -257,6 +272,14 @@ FunctionOverloadResolverPtr UserDefinedExecutableFunctionFactory::tryGet(const S
     {
         auto executable_function = std::static_pointer_cast<const UserDefinedExecutableFunction>(load_result.object);
         auto function = std::make_shared<UserDefinedFunction>(std::move(executable_function), std::move(context), std::move(parameters));
+
+        if (CurrentThread::isInitialized())
+        {
+            auto query_context = CurrentThread::get().getQueryContext();
+            if (query_context && query_context->getSettingsRef()[Setting::log_queries])
+                query_context->addQueryFactoriesInfo(Context::QueryLogFactories::Function, function_name);
+        }
+
         return std::make_unique<FunctionToOverloadResolverAdaptor>(std::move(function));
     }
 

tests/integration/test_executable_udf_names_in_used_functions_column/__init__.py

@@ -0,0 +1,2 @@
+<clickhouse>
+</clickhouse>

tests/integration/test_executable_udf_names_in_used_functions_column/config/executable_user_defined_functions_config.xml

@@ -0,0 +1,45 @@
+<functions>
+    <function>
+        <type>executable</type>
+        <name>test_function_bash</name>
+        <return_type>String</return_type>
+        <argument>
+            <type>String</type>
+        </argument>
+        <format>TabSeparated</format>
+        <command>input.sh</command>
+    </function>
+
+    <function>
+        <type>executable_pool</type>
+        <name>test_function_pool_bash</name>
+        <return_type>String</return_type>
+        <argument>
+            <type>String</type>
+        </argument>
+        <format>TabSeparated</format>
+        <command>input.sh</command>
+    </function>
+
+    <function>
+        <type>executable</type>
+        <name>test_function_python</name>
+        <return_type>String</return_type>
+        <argument>
+            <type>String</type>
+        </argument>
+        <format>TabSeparated</format>
+        <command>input.py</command>
+    </function>
+
+    <function>
+        <type>executable_pool</type>
+        <name>test_function_pool_python</name>
+        <return_type>String</return_type>
+        <argument>
+            <type>String</type>
+        </argument>
+        <format>TabSeparated</format>
+        <command>input.py</command>
+    </function>
+</functions>

tests/integration/test_executable_udf_names_in_used_functions_column/functions/test_function_config.xml

@@ -0,0 +1,100 @@
+import os
+import sys
+import time
+import uuid
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
+
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+node = cluster.add_instance("node", stay_alive=True, main_configs=[])
+
+
+def skip_test_msan(instance):
+    if instance.is_built_with_memory_sanitizer():
+        pytest.skip("Memory Sanitizer cannot work with vfork")
+
+
+def copy_file_to_container(local_path, dist_path, container_id):
+    os.system(
+        "docker cp {local} {cont_id}:{dist}".format(
+            local=local_path, cont_id=container_id, dist=dist_path
+        )
+    )
+
+
+config = """<clickhouse>
+    <user_defined_executable_functions_config>/etc/clickhouse-server/functions/test_function_config.xml</user_defined_executable_functions_config>
+</clickhouse>"""
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+
+        node.replace_config(
+            "/etc/clickhouse-server/config.d/executable_user_defined_functions_config.xml",
+            config,
+        )
+
+        copy_file_to_container(
+            os.path.join(SCRIPT_DIR, "functions/."),
+            "/etc/clickhouse-server/functions",
+            node.docker_id,
+        )
+        copy_file_to_container(
+            os.path.join(SCRIPT_DIR, "user_scripts/."),
+            "/var/lib/clickhouse/user_scripts",
+            node.docker_id,
+        )
+
+        node.restart_clickhouse()
+
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def check_executable_udf_functions(functions):
+    if not functions:
+        assert False
+    execute_udf = "SELECT {}" if len(functions) == 1 else "SELECT concat({})"
+    execute_udf = execute_udf.format(','.join(map(lambda function : "{}(1)".format(function), functions)))
+    query_id = uuid.uuid4().hex
+    node.query(execute_udf, query_id=query_id)
+    node.query("SYSTEM FLUSH LOGS")
+
+    used_functions = set(map(repr, functions))
+    if len(functions) > 1:
+        used_functions.add("\'concat\'")
+
+    query_result = node.query("SELECT used_functions FROM system.query_log "
+                              "WHERE type = \'QueryFinish\' AND query_id = \'{}\'".format(query_id))
+    query_result = set(query_result[1:len(query_result) - 2].split(','))
+    assert query_result == used_functions
+
+
+def test_executable_function_single(started_cluster):
+    skip_test_msan(node)
+    check_executable_udf_functions(["test_function_bash"])
+
+
+def test_executable_function_multiple(started_cluster):
+    skip_test_msan(node)
+    check_executable_udf_functions(["test_function_bash", "test_function_python"])
+
+
+def test_executable_function_multiple_pool(started_cluster):
+    skip_test_msan(node)
+    check_executable_udf_functions(["test_function_bash", "test_function_python", "test_function_pool_bash", "test_function_pool_python"])
+
+
+def test_executable_function_duplicate(started_cluster):
+    skip_test_msan(node)
+    check_executable_udf_functions(["test_function_bash", "test_function_bash"])

tests/integration/test_executable_udf_names_in_used_functions_column/test.py

@@ -0,0 +1,8 @@
+#!/usr/bin/python3
+
+import sys
+
+if __name__ == "__main__":
+    for line in sys.stdin:
+        print(line, end="")
+        sys.stdout.flush()

tests/integration/test_executable_udf_names_in_used_functions_column/user_scripts/input.py

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+while read read_data;
+    do printf "$read_data\n";
+done