Support distributed processing by pickling expressions#1543

Draft

timsaucer wants to merge 57 commits into

apache:mainfrom

timsaucer:feat/expr-pickle

Commits on May 15, 2026

feat: pickle support for Expr via PythonLogicalCodec inline encoding

timsaucer
and
claude
committed
docs(pickle): user guide + Ray example + CI backstop + UDF.name

timsaucer
and
claude
committed
refactor: drop dead state from PythonFunctionScalarUDF

timsaucer
and
claude
committed
refactor(codec): use arrow-rs native IPC for schema serialization

timsaucer
and
claude
committed
docs: strip implementation jargon from Expr pickle docstrings

timsaucer
and
claude
committed
docs: reframe distributed-expression docs around the user goal

timsaucer
and
claude
committed
feat: inline encoding for Python window UDFs

timsaucer
and
claude
committed
feat: inline encoding for Python aggregate UDFs

timsaucer
and
claude
committed
docs: extend the inline-UDF guarantee to aggregate + window UDFs

timsaucer
and
claude
committed
docs(distributing-expressions): link to pickle docs, generalize UDF kinds

timsaucer
and
claude
committed
docs: drop manual pickle.dumps/loads from worker examples

timsaucer
and
claude
committed
docs: restructure distribution page for the multiple-approach story

timsaucer
and
claude
committed
chore: untrack internal design doc that was committed by accident

timsaucer
and
claude
committed
refactor: rename MultiColumnWindowUDF -> PythonFunctionWindowUDF

timsaucer
and
claude
committed
docs(distributing-work): drop __main__ guard from example to match site style

timsaucer
and
claude
committed
feat: per-session toggle for Python UDF inline encoding

timsaucer
and
claude
committed
docs: pickle module security warning link + move user-facing prose

timsaucer
and
claude
committed
docs(ray-example): drop unnecessary UDF registration

timsaucer
and
claude
committed
feat: sender-side session context for outbound pickles

timsaucer
and
claude
committed
fix(ipc): doctest collection failure on module docstring

timsaucer
and
claude
committed
docs(example): multiprocessing pool with distinct expressions per worker

timsaucer
and
claude
committed
fix(codec): preserve UDAF state field metadata across pickle round-trip

timsaucer
and
claude
committed
refactor(codec): pin Volatility wire format to explicit match

timsaucer
and
claude
committed
fix: drop Python callable from UDF Hash impls

timsaucer
and
claude
committed
perf(codec): cache cloudpickle module handle in a OnceLock

timsaucer
and
claude
committed
refactor(udf): scalar from_parts now takes Vec<DataType>, matching udaf/udwf

timsaucer
and
claude
committed
docs(pyproject): cloudpickle covers aggregate and window UDFs too

timsaucer
and
claude
committed
refactor(codec): drop redundant buf.is_empty() guards

timsaucer
and
claude
committed
refactor(expr): drop redundant bytes(...) wrap in to_bytes

timsaucer
and
claude
committed
refactor(tests): drop redundant inline Expr reimports

timsaucer
and
claude
committed
docs(codec): justify synthetic input field names and nullability

timsaucer
and
claude
committed
fix(codec): use Execution error for inline-payload refusal

timsaucer
and
claude
committed
refactor(codec): cache cloudpickle via per-interpreter PyOnceLock

timsaucer
and
claude
committed
test(pickle): probe multiprocessing lazily via cached fixture

timsaucer
and
claude
committed
refactor(udwf): drop redundant PythonFunctionWindowUDF::from_parts

timsaucer
and
claude
committed
refactor(udaf): name synthesized state fields state_{i}

timsaucer
and
claude
committed
perf(udf): pointer-identity fast path in PartialEq for Python UDFs

timsaucer
and
claude
committed
docs(example): note null→0 coercion in above-threshold UDF

timsaucer
and
claude
committed
test(pickle): tighten refusal assertion from Exception to RuntimeError

timsaucer
and
claude
committed
docs(codec): clarify strict-mode security scope

timsaucer
and
claude
committed
build(uv): regenerate uv.lock with current uv

timsaucer
and
claude
committed
test(pickle): rename strict-pickle inline-worker test for accuracy

timsaucer
and
claude
committed
docs(distributing): note fork-worker threading.local() inheritance

timsaucer
and
claude
committed
test(ffi): strict-mode Expr round-trip with an FFI-capsule scalar UDF

timsaucer
and
claude
committed
fix(udf): name property reports FFI capsule name, not the ignored ctor arg

timsaucer
and
claude
committed
refactor(ctx): make with_python_udf_inlining `enabled` keyword-only

timsaucer
and
claude
committed
refactor(codec): factor shared schema/signature helpers out of UDF encoders

timsaucer
and
claude
committed
feat(codec): split UDF wire-format magic into family prefix + version byte

timsaucer
and
claude
committed
fix(codec): point strict refusal at the sender, not the receiver

timsaucer
and
claude
committed
fix(udf): log Python __eq__ exceptions instead of silently mapping to unequal

timsaucer
and
claude
committed
docs(expr): restate to_bytes behavior when ctx is None

timsaucer
and
claude
committed
docs(ipc): explain why _resolve_ctx imports SessionContext lazily

timsaucer
and
claude
committed
chore: replace blanket S301 allow with per-line noqa at pickle.loads

timsaucer
and
claude
committed
test(pickle): drop dead register_udf in strict refusal test

timsaucer
and
claude
committed
test(pickle): cover with_python_udf_inlining toggle off-then-on

timsaucer
and
claude
committed
docs(distributing): narrow Python 3.14 start-method note to Linux

timsaucer
and
claude
committed
docs: centralize pickle security caveat, link from codec/context docs

timsaucer
and
claude
committed