WebAssembly User Defined Functions

<clickhouse>
    <allow_experimental_webassembly_udf>true</allow_experimental_webassembly_udf>
    <webassembly_udf_engine>wasmtime</webassembly_udf_engine>
</clickhouse>

cat << 'EOF' | wasm-tools parse | clickhouse client -q "INSERT INTO system.webassembly_modules (name, code) SELECT 'collatz', code FROM input('code String') FORMAT RawBlob"
(module
  (func $next (param $n i32) (result i32)
    local.get $n i32.const 1 i32.and
    (if (result i32)
      (then local.get $n i32.const 3 i32.mul i32.const 1 i32.add)
      (else local.get $n i32.const 2 i32.div_u)))
  (func $steps (export "steps") (param $n i32) (result i32)
    (local $count i32)
    local.get $n i32.const 1 i32.lt_u
    (if (then i32.const 0 return))
    (block $done (loop $loop
      local.get $n i32.const 1 i32.eq br_if $done
      local.get $n call $next local.set $n
      local.get $count i32.const 1 i32.add local.set $count
      br $loop))
    local.get $count)
)
EOF

CREATE FUNCTION collatz_steps LANGUAGE WASM ARGUMENTS (n UInt32) RETURNS UInt32 FROM 'collatz' :: 'steps';

SELECT groupArray(collatz_steps(number :: UInt32))
FROM numbers(1, 100)
FORMAT TSV

[0,1,7,2,5,8,16,3,19,6,14,9,9,17,17,4,12,20,20,7,7,15,15,10,23,10,111,18,18,18,106,5,26,13,13,21,21,21,34,8,109,8,29,16,16,16,104,11,24,24,24,11,11,112,112,19,32,19,32,19,19,107,107,6,27,27,27,14,14,14,102,22,115,22,14,22,22,35,35,9,22,110,110,9,9,30,30,17,30,17,92,17,17,105,105,12,118,25,25,25]

INSERT INTO system.webassembly_modules (name, code)
SELECT 'my_module', base64Decode('AGFzbQEAAAA...');

INSERT INTO system.webassembly_modules (name, code, hash)
SELECT 'my_module', base64Decode('...'), reinterpretAsUInt256(unhex('369f...c57d'));

Code: 674. DB::Exception: WebAssembly module 'collatz' not found:
while adding user defined function `collatz_steps`. (RESOURCE_NOT_FOUND)

cat collatz.wasm | clickhouse client -q "
  INSERT INTO FUNCTION cluster('default', 'system', 'webassembly_modules') (name, code)
  SELECT 'collatz', code FROM input('code String') FORMAT RawBlob"

CREATE FUNCTION collatz_steps ON CLUSTER 'default'
LANGUAGE WASM FROM 'collatz' :: 'steps'
ARGUMENTS (n UInt32) RETURNS UInt32;

SELECT hostName(), name FROM clusterAllReplicas('default', system.webassembly_modules) WHERE name = 'collatz';

SELECT name, lower(hex(reinterpretAsFixedString(hash))) AS sha256 FROM system.webassembly_modules

   ┌─name────┬─sha256───────────────────────────────────────────────────────────┐
1. │ collatz │ a084a10b7b5cb07db198bc93bf1f3c1f8cb8ef279df7a4f6b66b1cdd55d79c48 │
   └─────────┴──────────────────────────────────────────────────────────────────┘

DELETE FROM system.webassembly_modules WHERE name = 'collatz';

-- Bulk-delete every module whose name starts with `tmp_` (literal underscore is escaped as `\_`):
DELETE FROM system.webassembly_modules WHERE name LIKE 'tmp\_%';

CREATE [OR REPLACE] FUNCTION function_name
LANGUAGE WASM
FROM 'module_name' [:: 'source_function_name']
ARGUMENTS ( [name type[, ...]] | [type[, ...]] )
RETURNS return_type
[ABI ROW_DIRECT | ABI BUFFERED_V1 | ABI ASSEMBLYSCRIPT]
[DETERMINISTIC]
[SHA256_HASH 'hex']
[SETTINGS key = value[, ...]];

(func (param i32 i64 f32) (result f64) ...)

CREATE FUNCTION my_func ARGUMENTS (Int32, UInt64, Float32) RETURNS Float64 ...

(module
  ;; Allocate a new buffer of specified size
  ;; Returns: handle to Buffer structure (not direct data pointer!) with pointer to data and size
  (func (export "clickhouse_create_buffer")
    (param $size i32)    ;; Size of data to allocate
    (result i32))        ;; Returns buffer handle with enough space

  ;; Free a buffer by its handle
  (func (export "clickhouse_destroy_buffer")
    (param $handle i32)  ;; Buffer handle to free
    (result))            ;; No return value

    ;; User-defined function
    (func (export "user_defined_function1")
      (param $input_buffer_handle i32)  ;; Input buffer handle
      (param $n i32)                    ;; Number of rows in input
      (result i32))                     ;; Returns output buffer handle
)

#include <stdint.h>

typedef struct {
    uint8_t * data;
    uint32_t size;
} ClickhouseBuffer;

#define HEAP_SIZE (1 << 24)
static _Alignas(16) uint8_t heap[HEAP_SIZE];
static uint32_t heap_pos = 0;
static uint32_t live_buffers = 0;

__attribute__((export_name("clickhouse_create_buffer")))
ClickhouseBuffer * clickhouse_create_buffer(uint32_t size)
{
    uint32_t total = (sizeof(ClickhouseBuffer) + size + 15u) & ~15u;
    if (heap_pos + total > HEAP_SIZE)
        return 0; /* a zero handle makes the host fail the query with an error */
    ClickhouseBuffer * buf = (ClickhouseBuffer *) &heap[heap_pos];
    buf->data = &heap[heap_pos + sizeof(ClickhouseBuffer)];
    buf->size = size;
    heap_pos += total;
    ++live_buffers;
    return buf;
}

__attribute__((export_name("clickhouse_destroy_buffer")))
void clickhouse_destroy_buffer(ClickhouseBuffer * buf)
{
    if (--live_buffers == 0)
        heap_pos = 0;
}

/* RowBinary prefixes each String with its byte length as an unsigned varint (LEB128) */
static uint64_t read_varint(const uint8_t ** pp)
{
    uint64_t value = 0;
    for (int shift = 0;; shift += 7)
    {
        uint8_t b = *(*pp)++;
        value |= (uint64_t)(b & 0x7f) << shift;
        if (!(b & 0x80))
            return value;
    }
}

static void write_varint(uint8_t ** pp, uint64_t value)
{
    while (value >= 0x80)
    {
        *(*pp)++ = (uint8_t)(value | 0x80);
        value >>= 7;
    }
    *(*pp)++ = (uint8_t)value;
}

/* Reverses the bytes of each input string */
__attribute__((export_name("str_reverse")))
ClickhouseBuffer * str_reverse(ClickhouseBuffer * input, uint32_t num_rows)
{
    ClickhouseBuffer * out = clickhouse_create_buffer(input->size);
    if (!out)
        return 0;
    const uint8_t * in = input->data;
    uint8_t * o = out->data;
    for (uint32_t row = 0; row < num_rows; ++row)
    {
        uint64_t len = read_varint(&in);
        write_varint(&o, len);
        for (uint64_t i = 0; i < len; ++i)
            o[i] = in[len - 1 - i];
        in += len;
        o += len;
    }
    out->size = (uint32_t)(o - out->data);
    return out;
}

clang --target=wasm32 -ffreestanding -nostdlib -fno-builtin -c str_reverse.c
wasm-ld --no-entry str_reverse.o -o str_reverse.wasm

cat str_reverse.wasm | clickhouse client -q "INSERT INTO system.webassembly_modules (name, code) SELECT 'str_reverse', code FROM input('code String') FORMAT RawBlob"

CREATE FUNCTION str_reverse LANGUAGE WASM ABI BUFFERED_V1
FROM 'str_reverse' :: 'str_reverse'
ARGUMENTS (s String) RETURNS String
SETTINGS serialization_format = 'RowBinary';

SELECT str_reverse(toString(number + 100)) FROM numbers(3);

001
101
201

asc src.ts --runtime incremental --exportRuntime -o src.wasm

// src.ts
export function add(a: u32, b: u32): u32 {
  return a + b;
}

export function greet(name: string): string {
  return "Hello, " + name + "!";
}

CREATE FUNCTION as_add
    LANGUAGE WASM ABI ASSEMBLYSCRIPT
    FROM 'as_example' :: 'add'
    ARGUMENTS (a UInt32, b UInt32) RETURNS UInt32;

CREATE FUNCTION as_greet
    LANGUAGE WASM ABI ASSEMBLYSCRIPT
    FROM 'as_example' :: 'greet'
    ARGUMENTS (name String) RETURNS String;

use clickhouse_wasm_udf_bindgen::clickhouse_udf;

#[clickhouse_udf]
pub fn some_udf(data: String) -> HashMap<String, String> {
    // Your implementation here
}

SET webassembly_udf_max_fuel = 200000;
SELECT my_wasm_udf(column) FROM table;