Add list_transform expression

Adds a ListTransform scalar fn implementing DuckDB-style list_transform(l,
lambda x: ...) as a Vortex expression. The lambda body is an ordinary
Expression stored in the fn's options, evaluated with the list's elements
array as its root scope (root() inside the body refers to the element).
Execution rewraps the list structure around a deferred apply() of the body
to the elements child, so offsets and list validity pass through untouched
and no element values are computed eagerly.

Includes:
- identity (body == root) and fusion (transform-of-transform) rewrites in
  simplify_untyped, plus list_length(list_transform(l, f)) -> list_length(l)
- proto serde of the options-embedded body via the session registry
- fallibility delegation to the body, since generic tree walks cannot see
  options-embedded expressions
- design doc under scalar_fn/fns/list_transform/design.md

Signed-off-by: Matt Katz <mhkatz97@gmail.com>

Author: mhk197

vortex-array/src/expr/exprs.rs

@@ -38,6 +38,8 @@ use crate::scalar_fn::fns::like::Like;
 use crate::scalar_fn::fns::like::LikeOptions;
 use crate::scalar_fn::fns::list_contains::ListContains;
 use crate::scalar_fn::fns::list_length::ListLength;
+use crate::scalar_fn::fns::list_transform::ListTransform;
+use crate::scalar_fn::fns::list_transform::ListTransformOptions;
 use crate::scalar_fn::fns::literal::Literal;
 use crate::scalar_fn::fns::mask::Mask;
 use crate::scalar_fn::fns::merge::DuplicateHandling;
@@ -765,3 +767,21 @@ pub fn ext_storage(input: Expression) -> Expression {
 pub fn list_length(input: Expression) -> Expression {
     ListLength.new_expr(EmptyOptions, [input])
 }
+
+// ---- ListTransform ----
+
+/// Creates an expression that transforms every element of a list with `body`, preserving the
+/// list's structure (offsets and list-level validity). This is akin to DuckDB's
+/// `list_transform()` with a lambda.
+///
+/// `body` is evaluated with the list's elements as its root scope: within `body`, `root()`
+/// refers to the element rather than the enclosing row. For example, DuckDB's
+/// `list_transform(tags, lambda x: x + 1)` is expressed as:
+///
+/// ```rust
+/// # use vortex_array::expr::{checked_add, get_item, list_transform, lit, root};
+/// let expr = list_transform(get_item("tags", root()), checked_add(root(), lit(1)));
+/// ```
+pub fn list_transform(input: Expression, body: Expression) -> Expression {
+    ListTransform.new_expr(ListTransformOptions { body }, [input])
+}

vortex-array/src/scalar_fn/fns/list_length.rs

@@ -31,6 +31,7 @@ use crate::scalar_fn::EmptyOptions;
 use crate::scalar_fn::ExecutionArgs;
 use crate::scalar_fn::ScalarFnId;
 use crate::scalar_fn::ScalarFnVTable;
+use crate::scalar_fn::fns::list_transform::ListTransform;
 use crate::scalar_fn::fns::operators::Operator;
 
 /// Number of elements in each list of a `List` or `FixedSizeList` typed array.
@@ -98,6 +99,20 @@ impl ScalarFnVTable for ListLength {
         list_length(&input, nullability, ctx)
     }
 
+    fn simplify_untyped(
+        &self,
+        _options: &Self::Options,
+        expr: &Expression,
+    ) -> VortexResult<Option<Expression>> {
+        // `list_transform` preserves list lengths, so skip it entirely:
+        // list_length(list_transform(l, f)) == list_length(l).
+        let input = expr.child(0);
+        if input.is::<ListTransform>() {
+            return Ok(Some(expr.clone().with_children([input.child(0).clone()])?));
+        }
+        Ok(None)
+    }
+
     fn validity(
         &self,
         _: &Self::Options,
@@ -173,7 +188,7 @@ fn list_length_from_offsets(list: ArrayView<'_, List>) -> VortexResult<ArrayRef>
 }
 
 /// Matches an `Array<List>`, `Array<ListView>`, or `Array<FixedSizeList>`
-struct AnyList;
+pub(crate) struct AnyList;
 
 impl Matcher for AnyList {
     type Match<'a> = ();