Implement function materialization

dbt/adapters/clickhouse/__version__.py

@@ -1 +1 @@
-version = '1.10.0'
+version = "1.11.0"

dbt/adapters/clickhouse/connections.py

@@ -3,10 +3,15 @@
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, Optional, Tuple, Union
 
+from dbt_common.events.contextvars import get_node_info
+from dbt_common.events.functions import fire_event
+from dbt_common.utils import cast_to_str
+
 import dbt.exceptions
 from dbt.adapters.clickhouse.dbclient import ChRetryableException, get_db_client
 from dbt.adapters.clickhouse.logger import logger
 from dbt.adapters.contracts.connection import AdapterResponse, Connection
+from dbt.adapters.events.types import SQLQuery
 from dbt.adapters.sql import SQLConnectionManager
 
 if TYPE_CHECKING:
@@ -87,7 +92,14 @@ def execute(
         client = conn.handle
 
         with self.exception_handler(sql):
-            logger.debug(f'On {conn.name}: {sql}...')
+            fire_event(
+                SQLQuery(
+                    conn_name=cast_to_str(conn.name),
+                    node_info=get_node_info(),
+                    sql=sql,
+                )
+            )
+            logger.debug(f"On {conn.name}: {sql}...")
             pre = time.time()
             if fetch:
                 query_result = client.query(sql)
@@ -116,7 +128,20 @@ def add_query(
         conn = self.get_thread_connection()
         client = conn.handle
         with self.exception_handler(sql):
-            logger.debug(f'On {conn.name}: {sql}...')
+            if abridge_sql_log:
+                log_sql = "{}...".format(sql[:512])
+            else:
+                log_sql = sql
+
+            fire_event(
+                SQLQuery(
+                    conn_name=cast_to_str(conn.name),
+                    node_info=get_node_info(),
+                    sql=log_sql,
+                )
+            )
+
+            logger.debug(f"On {conn.name}: {sql}...")
             pre = time.time()
             client.command(sql)
             status = self.get_status(client)

dbt/adapters/clickhouse/relation.py

@@ -1,14 +1,16 @@
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Type
 
-from dbt.adapters.base.relation import BaseRelation, EventTimeFilter, Path, Policy, Self
-from dbt.adapters.clickhouse.query import quote_identifier
-from dbt.adapters.contracts.relation import HasQuoting, RelationConfig
 from dbt_common.dataclass_schema import StrEnum
 from dbt_common.exceptions import DbtRuntimeError
 from dbt_common.utils import deep_merge
 
-NODE_TYPE_SOURCE = 'source'
+from dbt.adapters.base.relation import BaseRelation, EventTimeFilter, Path, Policy, Self
+from dbt.adapters.clickhouse.query import quote_identifier
+from dbt.adapters.contracts.relation import HasQuoting, RelationConfig
+
+NODE_TYPE_SOURCE = "source"
+NODE_TYPE_FUNCTION = "function"
 
 
 @dataclass
@@ -33,6 +35,7 @@ class ClickHouseRelationType(StrEnum):
     External = "external"
     Ephemeral = "ephemeral"
     Dictionary = "dictionary"
+    Function = "function"
 
 
 @dataclass(frozen=True, eq=False, repr=False)
@@ -146,6 +149,9 @@ def create_from(
         if relation_config.resource_type == NODE_TYPE_SOURCE:
             if schema == relation_config.source_name and relation_config.database:
                 schema = relation_config.database
+        elif relation_config.resource_type == NODE_TYPE_FUNCTION:
+            # ClickHouse functions are global and don't belong to a schema, so we ignore the schema.
+            schema = None
         else:
             # quoting is only available for non-source nodes
             cluster = quoting.credentials.cluster or ""

dbt/include/clickhouse/macros/materializations/functions/scalar.sql

@@ -0,0 +1,15 @@
+{% macro clickhouse__formatted_scalar_function_args_sql() %}
+    {% set args = [] %}
+    {% for arg in model.arguments -%}
+      {%- do args.append(arg.name) -%}
+    {%- endfor %}
+    {{ args | join(', ') }}
+{% endmacro %}
+
+{% macro clickhouse__scalar_function_create_replace_signature_sql(target_relation) %}
+  CREATE OR REPLACE FUNCTION {{ target_relation.include(database=false, schema=false) }} {{ on_cluster_clause(this) }} AS ({{ clickhouse__formatted_scalar_function_args_sql() }}) ->
+{% endmacro %}
+
+{% macro clickhouse__scalar_function_body_sql() %}
+  {{ model.compiled_code }}
+{% endmacro %}

dev_requirements.txt

@@ -1,4 +1,4 @@
-dbt-core>=1.10.0,<1.11
+dbt-core>=1.11.0
 dbt-tests-adapter>=1.10,<2.0
 pytest>=7.2.0
 pytest-dotenv==0.5.2

tests/integration/adapter/udf/test_udf.py

@@ -0,0 +1,23 @@
+"""
+test UDF creation support for dbt-clickhouse
+"""
+
+import pytest
+
+import dbt.tests.adapter.functions.files as files
+from dbt.tests.adapter.functions.test_udfs import UDFsBasic
+
+# we'll import helper used by the core test project to write directories
+
+MY_UDF_SQL = """
+price * 2
+""".strip()
+
+
+class TestUDFBasics(UDFsBasic):
+    @pytest.fixture(scope="class")
+    def functions(self):
+        return {
+            "price_for_xlarge.sql": MY_UDF_SQL,
+            "price_for_xlarge.yml": files.MY_UDF_YML,
+        }