feat(udf): native Rust scalar UDFs

crates/krishiv-common/krishiv-error-classes.json

@@ -113,6 +113,12 @@
     ],
     "sqlState" : "42K0I"
   },
+  "KRISHIV_UDF_NOT_AVAILABLE" : {
+    "message" : [
+      "The user-defined function <name>@<version> is not available on this worker: register it on the executing session before running the plan."
+    ],
+    "sqlState" : "39000"
+  },
   "KRISHIV_UNKNOWN_FORMAT_OPTION" : {
     "message" : [
       "The option <option> is not an option of the <format> data source.<suggestion> Remove it, or set \"krishiv.io.strictOptions\" to \"false\" to ignore unknown options with a warning."

crates/krishiv-exec/Cargo.toml

@@ -29,6 +29,7 @@ krishiv-memory = { path = "../krishiv-memory" }
 krishiv-optimizer = { path = "../krishiv-optimizer" }
 krishiv-plan = { path = "../krishiv-plan" }
 krishiv-types = { path = "../krishiv-types" }
+krishiv-udf-sdk = { path = "../krishiv-udf-sdk" }
 thiserror.workspace = true
 # The query profile's JSON document (`profile`).
 serde_json = { workspace = true, features = ["std"] }
@@ -56,6 +57,10 @@ loom = { workspace = true, features = ["futures"] }
 # Row-layout key-page prototype (P2-T157); not part of the default build.
 row-table-prototype = []
 
+[[bench]]
+name = "udf"
+harness = false
+
 [[bench]]
 name = "pipeline"
 harness = false

crates/krishiv-exec/benches/udf.rs

@@ -0,0 +1,129 @@
+// SPDX-License-Identifier: Apache-2.0
+//! An identity UDF against the built-in it mirrors: invoking each over
+//! 8,192-row batches, through the same [`ScalarFunction`] interface the
+//! evaluator uses. The UDF's per-batch cost over `abs` is the wrapper
+//! (checks, null masks, unwind boundary): the identity itself copies its
+//! input in both cases.
+
+#![expect(
+    missing_docs,
+    reason = "criterion_group! generates an undocumented function"
+)]
+
+use std::hint::black_box;
+use std::sync::Arc;
+
+use arrow_array::{ArrayRef, Int32Array};
+use arrow_schema::DataType;
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use krishiv_functions::scalar::{Arg, BindArg, BindOptions, ScalarFunction, bind_scalar, bind_udf};
+use krishiv_types::DataType as SparkType;
+use krishiv_udf_sdk::{
+    ArgField, InvokeContext, ReturnField, ScalarUdf, Signature, UdfArg, UdfError, UdfInputType,
+};
+
+const ROWS: usize = 8_192;
+
+/// The identity over `INT`.
+#[derive(Debug)]
+struct Identity;
+
+impl ScalarUdf for Identity {
+    fn name(&self) -> &str {
+        "identity_i32"
+    }
+
+    fn version(&self) -> u32 {
+        1
+    }
+
+    fn signature(&self) -> &Signature {
+        use std::sync::OnceLock;
+        static SIG: OnceLock<Signature> = OnceLock::new();
+        SIG.get_or_init(|| Signature::Exact(vec![UdfInputType::Exact(SparkType::Integer)]))
+    }
+
+    fn return_field(&self, _: &[ArgField]) -> Result<ReturnField, UdfError> {
+        Ok(ReturnField {
+            data_type: SparkType::Integer,
+            nullable: false,
+        })
+    }
+
+    fn invoke(
+        &self,
+        args: &[UdfArg<'_>],
+        _: usize,
+        _: &InvokeContext,
+    ) -> Result<ArrayRef, UdfError> {
+        Ok(args[0].array())
+    }
+}
+
+fn batch() -> ArrayRef {
+    // Negative and positive values, so `abs` computes on every row.
+    Arc::new(Int32Array::from_iter_values(
+        (0..ROWS).map(|i| i32::try_from(i).unwrap_or(0).wrapping_sub(4096)),
+    ))
+}
+
+#[expect(
+    clippy::expect_used,
+    reason = "a broken fixture must fail the bench loudly, not skip it"
+)]
+fn bound_abs() -> Arc<dyn ScalarFunction> {
+    bind_scalar(
+        "abs",
+        &[BindArg::Column {
+            ty: DataType::Int32,
+            nullable: true,
+        }],
+        &BindOptions::new(),
+    )
+    .expect("abs(int)")
+}
+
+#[expect(
+    clippy::expect_used,
+    reason = "a broken fixture must fail the bench loudly, not skip it"
+)]
+fn bound_identity() -> Arc<dyn ScalarFunction> {
+    let udf: Arc<dyn ScalarUdf> = Arc::new(Identity);
+    bind_udf(
+        &udf,
+        &[DataType::Int32],
+        &SparkType::Integer,
+        false,
+        true,
+        None,
+    )
+    .expect("identity")
+}
+
+#[expect(
+    clippy::expect_used,
+    reason = "a failed invocation must fail the bench loudly, not skip it"
+)]
+fn bench(c: &mut Criterion) {
+    let mut group = c.benchmark_group("udf");
+    group.throughput(Throughput::Elements(ROWS as u64));
+    let input = batch();
+    for (id, bound) in [("abs", bound_abs()), ("identity_udf", bound_identity())] {
+        group.bench_with_input(
+            BenchmarkId::new("invoke_8192_rows", id),
+            &bound,
+            |b, bound: &Arc<dyn ScalarFunction>| {
+                b.iter(|| {
+                    let out = bound
+                        .invoke(black_box(&[Arg::Array(Arc::clone(&input))]), ROWS)
+                        .expect("invoke");
+                    black_box(out.len())
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
+criterion_group!(benches, bench);
+criterion_main!(benches);

crates/krishiv-exec/src/physical/instantiate.rs

@@ -47,6 +47,7 @@ use krishiv_io::{
 };
 use krishiv_memory::SpillScope;
 use krishiv_plan::{AttrId, Expr};
+use krishiv_udf_sdk::UdfLibrary;
 use object_store::path::Path;
 
 use super::{
@@ -88,13 +89,15 @@ pub trait SpillScopes: Send + Sync + fmt::Debug {
 }
 
 /// What a worker resolves a plan's references against: the tables its
-/// scans read, the relation store its batch splits resolve through, the
-/// spill scope of the query, the I/O context, formats and configuration a
-/// write builds its sink from, and the values of the plan. One per query
+/// scans read, the session UDFs its function calls invoke, the relation store
+/// its batch splits resolve through, the spill scope of the query, the I/O
+/// context, formats and configuration a write builds its sink from, and the
+/// values of the plan. One per query
 /// and worker.
 #[derive(Clone)]
 pub struct WorkerContext {
     tables: Arc<dyn TableProviderResolver>,
+    udfs: Option<Arc<UdfLibrary>>,
     relations: Option<Arc<krishiv_memory::RelationStore>>,
     spill: Option<SpillScope>,
     scopes: Option<Arc<dyn SpillScopes>>,
@@ -152,6 +155,7 @@ impl WorkerContext {
     pub fn new(tables: Arc<dyn TableProviderResolver>) -> Self {
         Self {
             tables,
+            udfs: None,
             relations: None,
             spill: None,
             scopes: None,
@@ -163,6 +167,14 @@ impl WorkerContext {
         }
     }
 
+    /// The same context invoking session UDFs from `udfs`. A plan calling
+    /// a UDF the context lacks fails with `KRISHIV_UDF_NOT_AVAILABLE`.
+    #[must_use]
+    pub fn with_udfs(mut self, udfs: Arc<UdfLibrary>) -> Self {
+        self.udfs = Some(udfs);
+        self
+    }
+
     /// The same context resolving a scan's batch splits (the
     /// [`ScanSplits::Batches`] of its task) through `relations`: the
     /// session's relation store.
@@ -245,6 +257,7 @@ impl fmt::Debug for WorkerContext {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("WorkerContext")
             .field("tables", &self.tables)
+            .field("udfs", &self.udfs.is_some())
             .field("relations", &self.relations.is_some())
             .field("spill", &self.spill.is_some())
             .field("spill_scopes", &self.scopes.is_some())
@@ -277,11 +290,14 @@ pub fn instantiate_task(
             (None, Some(scopes)) => Some(scopes.scope()?),
             (None, None) => None,
         };
+        let mut env = EvalEnv::from_plan(plan.env());
+        // The UDFs travel with the worker, never with the plan.
+        env.udfs = worker.udfs.clone();
         let mut cx = Instantiation {
             worker,
             task: task.clone(),
             spill,
-            env: EvalEnv::from_plan(plan.env()),
+            env,
             values: worker
                 .values
                 .clone()

crates/krishiv-exec/src/planner/build.rs

@@ -181,9 +181,13 @@ pub(crate) struct Planner<'a> {
 impl<'a> Planner<'a> {
     fn new(cx: &'a PlanningContext, root: &LogicalPlan) -> Self {
         let next_id = max_id(root).saturating_add(1).max(1 << 62);
+        let mut env = EvalEnv::new(&cx.config, Arc::clone(&cx.user));
+        // Planning validates UDF calls against the session's library; the
+        // implementation a plan runs comes from its worker.
+        env.udfs = cx.udfs.clone();
         Self {
             cx,
-            env: EvalEnv::new(&cx.config, Arc::clone(&cx.user)),
+            env,
             warnings: Vec::new(),
             ctes: HashMap::new(),
             consumed_hints: HashSet::new(),

crates/krishiv-exec/src/planner/eval/func.rs

@@ -10,12 +10,64 @@ use krishiv_functions::scalar::{Arg, BindArg, BindOptions, ScalarFunction, bind_
 use krishiv_kernels::cast::FixedOffset;
 use krishiv_kernels::map::DedupPolicy;
 use krishiv_plan::{Expr, ExprNode, ResolvedFunction};
+use krishiv_udf_sdk::ScalarUdf;
 
 use super::strings::to_view;
 use super::{Input, PhysExpr, Value, literal_array, unsupported_expr};
 use crate::error::Result;
 use crate::planner::build::{internal, unsupported};
 
+/// Binds a session UDF call: wraps the implementation (resolved from the
+/// worker by the caller) like a built-in binding, over the arguments the
+/// analyzer cast.
+pub(super) fn compile_udf(
+    udf: &Arc<dyn ScalarUdf>,
+    f: &ResolvedFunction,
+    args: Vec<Arc<dyn PhysExpr>>,
+    input: Input<'_>,
+    _e: &Expr,
+) -> Result<Arc<dyn PhysExpr>> {
+    let bind_args = f
+        .args
+        .iter()
+        .zip(&args)
+        .map(|(ir, compiled)| match &ir.node {
+            ExprNode::Literal(l) => literal_array(l).map(BindArg::Literal),
+            _ => Ok(BindArg::Column {
+                ty: compiled.data_type().clone(),
+                nullable: compiled.nullable(),
+            }),
+        })
+        .collect::<Result<Vec<_>>>()?;
+    let arg_types: Vec<ArrowType> = bind_args.iter().map(BindArg::data_type).collect();
+    // `bind_udf` reports user faults as `FAILED_EXECUTE_UDF` and engine
+    // faults as `INTERNAL_ERROR` (never `KRISHIV_UNSUPPORTED_FEATURE`,
+    // which has no planning fallback), so there is nothing to map here.
+    let function = krishiv_functions::scalar::bind_udf(
+        udf,
+        &arg_types,
+        &f.data_type,
+        f.nullable,
+        input.env.ansi,
+        input.env.zone,
+    )?;
+    let out = f.data_type.to_arrow();
+    // The binding returns plain carriers; Spark's `STRING` and `BINARY`
+    // are views here. Both derive from the analyzer's type, so they only
+    // ever match or convert to a view (by construction, no mismatch).
+    let view = matches!(
+        (function.return_type(), &out),
+        (ArrowType::Utf8, ArrowType::Utf8View) | (ArrowType::Binary, ArrowType::BinaryView)
+    );
+    Ok(Arc::new(Call {
+        function,
+        args,
+        out,
+        view,
+        nullable: f.nullable,
+    }))
+}
+
 pub(super) fn compile(
     name: &str,
     f: &ResolvedFunction,

crates/krishiv-exec/src/planner/eval/hof.rs

@@ -447,6 +447,7 @@ mod tests {
             kernels: KernelCtx::new(),
             query_start_micros: 0,
             user: Arc::from("test"),
+            udfs: None,
             map_key_dedup_last_win: false,
             seed_salt: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }