Implementation:ArroyoSystems Arroyo Threaded Udf
| Knowledge Sources | |
|---|---|
| Domains | Streaming, UDF, Python |
| Last Updated | 2026-02-08 08:00 GMT |
Overview
ThreadedUdfInterpreter manages the lifecycle of a Python UDF by running it on a dedicated thread with its own SubInterpreter, communicating via synchronous channels for task submission and result collection.
Description
The ThreadedUdfInterpreter struct provides the entry point for loading and executing Python UDFs within Arroyo's streaming engine. Its new method is async but internally spawns a dedicated OS thread for Python execution:
Initialization Flow:
- A dedicated thread is spawned that creates a SubInterpreter for GIL isolation
- The Python UDF source code (body) is parsed via the parse method, which:
- Loads the Arroyo UDF Python library (UDF_PY_LIB)
- Executes the user's Python code in the sub-interpreter
- Calls get_udfs() to discover @udf-annotated functions (exactly one must be present)
- Extracts the function name, argument types (NullableType), and return type via extract_type_info
- The parsed metadata is sent back to the main thread via a sync_channel
- A PythonUDF struct is returned containing the function's Signature, type information, and communication channels
Execution Flow: When the engine invokes the UDF, it sends Vec<ArrayRef> arguments via task_tx. The dedicated thread calls execute, which:
- Evaluates the Python function by name in the sub-interpreter
- Iterates over each row, converting Arrow values to Python objects via Converter::get_pyobject
- Skips calling the UDF for null arguments on non-nullable parameters (returns py.None() instead)
- Calls the Python function with a PyTuple of arguments
- Converts all results back to an Arrow array via Converter::build_array
- Sends the result back via result_tx
Type Signature Generation: The get_typesignature method generates DataFusion TypeSignature::OneOf by computing the Cartesian product of alternative types for each argument (e.g., Int64 accepts Int8 through UInt64; Float64 accepts Float32 and Float64).
Usage
Used when a user defines a Python UDF in their Arroyo SQL pipeline. The engine calls ThreadedUdfInterpreter::new(body) during program initialization, which returns a PythonUDF that is registered as a DataFusion UDF.
Code Reference
Source Location
- Repository: ArroyoSystems_Arroyo
- File: crates/arroyo-udf/arroyo-udf-python/src/threaded.rs
Signature
pub struct ThreadedUdfInterpreter {}
impl ThreadedUdfInterpreter {
pub async fn new(body: Arc<String>) -> anyhow::Result<PythonUDF>;
}
// PythonUDF (returned by new()) contains:
// - name: Arc<String>
// - task_tx: SyncSender<Vec<ArrayRef>>
// - result_rx: Arc<Mutex<Receiver<anyhow::Result<ArrayRef>>>>
// - definition: Arc<String>
// - signature: Arc<Signature>
// - arg_types: Arc<Vec<NullableType>>
// - return_type: Arc<NullableType>
Import
use arroyo_udf_python::threaded::ThreadedUdfInterpreter;
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| body | Arc<String> | Yes | Python source code containing exactly one @udf-annotated function |
| args | Vec<ArrayRef> | Yes (per invocation) | Arrow arrays representing the UDF arguments for a batch of rows |
Outputs
| Name | Type | Description |
|---|---|---|
| python_udf | PythonUDF | Initialized UDF handle with type signature, communication channels, and metadata |
| result | anyhow::Result<ArrayRef> | Arrow array of UDF results for the input batch, or an error |
Usage Examples
use std::sync::Arc;
use arroyo_udf_python::threaded::ThreadedUdfInterpreter;
// Python UDF source code
let body = Arc::new(r#"
from arroyo_udf import udf
@udf
def double(x: int) -> int:
return x * 2
"#.to_string());
// Initialize the UDF (spawns a dedicated thread with SubInterpreter)
let python_udf = ThreadedUdfInterpreter::new(body).await?;
// python_udf.name == "double"
// python_udf.signature contains TypeSignature for int argument
// Use task_tx/result_rx to invoke the UDF on Arrow batches