Implementation:ArroyoSystems Arroyo Sql Functions

Overview

SQL Functions registers Arroyo-specific scalar user-defined functions (UDFs) with the DataFusion function registry. These include JSON extraction functions, a multi-field hash function, JSON union serialization, and a helper for rewriting SQL function calls in the planning phase.

Description

The module provides:

• register_all: Registers all custom UDFs with a DataFusion FunctionRegistry:
  • get_first_json_object(json, path): Extracts the first matching value from a JSON string using a JSONPath expression. Returns Utf8.
  • extract_json(json, path): Extracts all matching values from a JSON string using a JSONPath expression. Returns List<Utf8>.
  • extract_json_string(json, path): Extracts and unquotes the first string value from a JSON string. Returns Utf8.
  • serialize_json_union(union): Serializes a sparse Arrow Union column to JSON string representation. Used for handling unstructured JSON data.
  • multi_hash(): A multi-field hash function used for key calculation in joins and aggregates.

• MultiHashFunction: Implements ScalarUDFImpl with a variadic signature accepting any combination of types. It uses Arrow's RowConverter to produce fixed-size binary hash keys.

• make_udf_function! macro: A helper macro for creating singleton UDF instances using OnceLock.

• rewrite_udf_calls: Rewrites SQL function calls from the schema provider's UDF registry. Handles both standard UDFs and async UDFs (with dylib configuration for native code execution).

Usage

register_all is called during schema provider initialization. The functions are then available in SQL queries.

Code Reference

Source Location

crates/arroyo-planner/src/functions.rs

Signature

pub fn register_all(registry: &mut dyn FunctionRegistry)

pub fn rewrite_udf_calls(
    expr: Expr,
    schema_provider: &ArroyoSchemaProvider,
) -> Result<(Expr, Option<DylibUdfConfig>)>

#[derive(Debug)]
pub struct MultiHashFunction {
    signature: Signature,
}

impl ScalarUDFImpl for MultiHashFunction {
    fn name(&self) -> &str
    fn signature(&self) -> &Signature
    fn return_type(&self, _: &[DataType]) -> Result<DataType>
    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue>
}

Import

use crate::functions::{register_all, rewrite_udf_calls, MultiHashFunction};

I/O Contract

Inputs

Name	Type	Description
registry	&mut dyn FunctionRegistry	DataFusion function registry to register UDFs with
json_string	Utf8	JSON string input for extraction functions
path	Utf8	JSONPath expression for extraction functions

Outputs

Name	Type	Description
extracted_value	Utf8	Extracted JSON value (from get_first_json_object, extract_json_string)
extracted_list	List<Utf8>	All matching JSON values (from extract_json)
hash_key	FixedSizeBinary(16)	16-byte hash key (from multi_hash)

Usage Examples

-- Extract a JSON field using JSONPath
SELECT get_first_json_object(payload, '$.user.name') as user_name
FROM events;

-- Extract all matching values
SELECT extract_json(payload, '$.items[*].price') as prices
FROM orders;

Related Pages

• ArroyoSystems_Arroyo_Table_Catalog - Schema provider that holds UDF definitions
• ArroyoSystems_Arroyo_Plan_Rewriter - Rewriter that invokes rewrite_udf_calls
• ArroyoSystems_Arroyo_Physical_Planner - Physical execution of UDF expressions