Python Arrow Expressions And UDFs

Datum has a lowerable Arrow expression tier and an opaque callable tier. The distinction is part of the API and the execution report:

• col(...), select(...), with_column(...), and expression filter(...) build typed kernel nodes. Datum validates their schemas during blueprint construction and lowers them to Arrow Rust kernels by default.
• map_batches(...) accepts arbitrary Python. It remains an opaque Arrow UDF and is never silently rewritten or rerouted as a lowerable expression.
• Source.map(...), Source.filter(...), and Source.flat_map(...) on integer streams are callable conveniences that wrap the callable in an Arrow UDF; arbitrary Rust closures likewise stay outside the implicit kernel-lowering contract.

Lowerable Expressions

Use datum.col(name) to build column references. Arithmetic operators create computed numeric expressions, comparisons create boolean expressions, and .cast(...) uses Arrow's safe cast semantics:

import datum
import pyarrow as pa

batch = pa.record_batch(
    [
        pa.array([1, 2, None, 4], type=pa.int64()),
        pa.array([10, 20, 30, 40], type=pa.int64()),
    ],
    names=["id", "offset"],
)

source = (
    datum.Source.from_arrow(batch)
    .with_column("total", datum.col("id") + datum.col("offset"))
    .with_column("id_f64", datum.col("id").cast(pa.float64()))
    .filter(datum.col("offset") >= 20)
    .select(["total", "id_f64"])
)

assert source.schema() == pa.schema(
    [
        pa.field("total", pa.int64(), nullable=True),
        pa.field("id_f64", pa.float64(), nullable=True),
    ]
)

with datum.Runtime() as runtime:
    table = source.run_collect(runtime)

The supported expression family is:

• numeric add, subtract, multiply, and divide, including typed scalar literals;
• equality and ordered comparisons over non-floating Arrow values;
• boolean-mask filters;
• safe Arrow casts;
• named projections and appended computed columns.

Column existence, operand types, cast support, filter boolean type, duplicate computed names, and input/output schemas are checked at the build call. No expression code runs while the blueprint is built. Null propagation and null-mask filtering are delegated to the Arrow kernels: a null in an arithmetic/comparison input produces a null output, and a null filter-mask entry is dropped.

Integer arithmetic is checked. Overflow and integer division by zero fail the materialized stream. Floating division follows Arrow's IEEE behavior, including infinities and NaN for division by zero. Floating comparisons are deliberately not lowerable today: arrow-rs comparison kernels use IEEE totalOrder, while Arrow C++ uses ordinary IEEE comparisons for signed zero and NaN. Datum rejects that expression at construction rather than allowing backend-dependent numeric drift; use an explicit map_batches(...) UDF when those semantics are required.

Reusable expression flows declare their input schema so validation remains eager:

flow = datum.BatchFlow.with_column(
    batch.schema,
    "total",
    datum.col("id") + datum.col("offset"),
).then_select(["total"])

source = datum.Source.from_arrow(batch).via(flow)

Kernel Backends And Reports

Every expression node accepts backend="arrow_rust", backend="python", or backend="auto". arrow_rust is the default and calls safe arrow-rs compute kernels. python compiles the same typed expression into direct pyarrow.compute calls backed by Arrow C++; it does not invoke a user callable. Reports expose both the requested and selected backend:

source = datum.Source.from_arrow(batch).with_column(
    "id_x2",
    datum.col("id") * 2,
    backend="auto",
)

node = source.execution_report().nodes[0]
assert node.lowerable is True
assert node.requested_backend == "auto"
assert node.selected_backend == "arrow_rust"
assert node.executor_tier == "arrow_batch_kernel"

Auto changes a family only when PyArrow/C++ beats Arrow Rust by more than 25% at two adjacent required batch sizes. The S1 arithmetic measurements found a PyArrow win only at 16 MiB; PyArrow lost at the adjacent 1 MiB size, so Auto currently selects Arrow Rust. An opaque map_batches node reports kind="opaque_python_udf", lowerable=False, executor_tier="python_udf", and no kernel backend.

Expression nodes are currently local-only. Datum Connect rejects a plan containing one instead of silently dropping or rerouting it; opaque map_batches UDF plans keep their existing Connect tier.

Per-node batch metrics

Metrics are off by default, so the plan-only report above performs no batch execution. Pass metrics_level="counts", "timing", or "stalls" to execute the stored source batches and return an immutable snapshot for each plan node:

report = source.execution_report(metrics_level="timing")
metric = report.node_metrics[0]

assert metric.batches_in == 1
assert metric.batches_out == 1
assert metric.rows_in == batch.num_rows
assert metric.rows_out == batch.num_rows
assert metric.processing_time_ns > 0

snapshot = report.metrics_snapshot()  # a fresh point-in-time copy

counts adds exact batch and row counts. timing additionally uses a monotonic clock around node kernel or callable work. stalls adds backpressure fields; synchronous batch-expression plans have zero stall time and no sampled queue depth. Calling an enabled execution report runs the batch plan, including any opaque Python UDF, so treat stateful UDF side effects exactly as you would for another materialization.

Opaque Python Callables

All user Python execution in Datum goes through Arrow UDF batches. There are two callable surfaces:

• Source.map(...), Source.filter(...), and Source.flat_map(...) on integer streams wrap your callable in a generated Arrow UDF. Datum sends one-column integer RecordBatch values through the Arrow boundary, then the wrapper loops over elements in Python.
• map_batches(...) is the vectorized tier. Your callable receives and returns whole pyarrow.RecordBatch values.

The map_batches(...) contract is:

pyarrow.RecordBatch -> pyarrow.RecordBatch

You declare the output schema when building the blueprint. Datum validates the first materialized output batch against that schema and turns Python exceptions into datum.StreamError.

Map Batches

import datum
import pyarrow as pa
import pyarrow.compute as pc

source_batch = pa.record_batch(
    [
        pa.array([1, 2, 3, 4], type=pa.int64()),
        pa.array([10, 20, 30, 40], type=pa.int64()),
    ],
    names=["id", "offset"],
)
output_schema = pa.schema([("total", pa.int64())])

def add_columns(batch: pa.RecordBatch) -> pa.RecordBatch:
    total = pc.add(batch.column(0), batch.column(1))
    return pa.record_batch([total], schema=output_schema)

with datum.Runtime() as runtime:
    table = (
        datum.Source.from_arrow(source_batch)
        .map_batches(add_columns, output_schema=output_schema)
        .run_collect(runtime)
    )

assert table.to_pydict() == {"total": [11, 22, 33, 44]}

Source.from_arrow() accepts PyArrow tables, record batches, record-batch readers, and iterables of record batches. Non-empty inputs infer their schema at construction. Empty inputs require schema=..., and declared schemas are checked against input batches before the source blueprint is returned. The tests also cover Polars and DuckDB Arrow capsule interop when those libraries are installed.

For reusable batch flows:

flow = datum.Flow.map_batches(add_columns, output_schema=output_schema)
source = datum.Source.from_arrow(source_batch).via(flow)

Schema And Tracebacks

Every Arrow batch edge has a known schema during blueprint construction. BatchSource.schema() and BatchFlow.schema() return the propagated schema, and each map_batches() call must declare output_schema=...:

project_schema = pa.schema([("id", pa.int64())])

def project_id(batch):
    return pa.record_batch([batch.column(0)], schema=project_schema)

stream = datum.Source.from_arrow(source_batch).map_batches(
    project_id,
    output_schema=project_schema,
)

assert stream.schema() == project_schema

If the UDF returns a batch with the wrong schema, the stream fails at the first bad output batch. If the UDF raises, the StreamError message includes the Python traceback text.

That is the deliberate execution boundary: Datum can validate and propagate the declared output_schema at build time, but arbitrary Python UDF code can still return a different RecordBatch only when it runs.

Copy Boundaries

Datum uses the Arrow PyCapsule path for the in-process batch boundary. The narrow zero-copy check in the benchmark record verifies that fixed-width identity and projection UDFs return buffers whose addresses match the source buffers; it does not mean every UDF is copy-free. Allocating new arrays, calling kernels that allocate outputs, or changing formats can still allocate.

The unavoidable visible unit is the batch. A Python RecordBatch object is constructed for the callable, so you want enough rows per batch to amortize that fixed crossing cost.

Performance Model

For a computed int64 column (left_x2 = left * 2), the S1 expression benchmark measured Arrow Rust at 223.72x wall / 223.99x CPU faster than an opaque scalar batch loop at 1 MiB, and 246.34x / 246.79x faster at 16 MiB. PyArrow/C++ was 27.17% faster than Arrow Rust at 16 MiB but 131.33% slower at 1 MiB, which is why the adjacent-size Auto rule keeps Arrow Rust selected. See the S1 benchmark ledger.

The remaining numbers in this section describe the opaque Python UDF crossing, not expression lowering.

All numbers here are from the PY-2 benchmark record and the lead-refined qualification table.

The final in-process identity/projection UDF table is functional but not performance-qualified: at 1,048,576 batch bytes, identity measured 281.7% overhead versus the pure Rust identity operator and projection measured 254.0%; at 16,777,216 batch bytes, identity measured 101.1% and projection measured 311.6%.

The refined compute UDF benchmark appends left_x2 = left * 2 with Arrow kernels:

Batch bytes	Rows/batch	Wall overhead vs Rust compute	CPU overhead vs Rust compute	Crossing constant us/batch
65,536	4,096	1151.0%	1155.2%	24.759
1,048,576	65,536	38.9%	38.8%	25.688
16,777,216	1,048,576	-56.1%	-56.1%	27.649

That is the practical model: there is a roughly batch-constant crossing cost, and useful work should be vectorized inside PyArrow, NumPy, Polars, or another Arrow-aware engine. The integer callable tier deliberately uses a Python row loop inside each Arrow batch for ergonomics; it is much slower than canned kernels and should be replaced with map_batches() or a built-in kernel when throughput matters.

GIL Behavior

Rust-only stream work detaches from Python. When Datum invokes your UDF, the worker thread attaches to Python and the callable runs under the GIL on normal GIL builds. Vectorized libraries may release the GIL internally, but Datum does not make arbitrary Python CPU code parallel in-process.

For parallel Python CPU work or dependency isolation, use Datum Connect UDF workers rather than in-process map_batches().