Ranked backlog: native scalar expressions with per-row optimization opportunities

[andygrove]

Follow-up to #4936. A scan of the native scalar expression implementations in datafusion-comet-spark-expr (native/spark-expr/) for per-row antipatterns: per-element builder loops that could be vectorized Arrow kernels, iter().map().collect() over Option/Result, per-row heap allocation (String/format!/to_string), per-row value(i) allocation, per-element MutableArrayData::extend, and constants recomputed inside the per-row loop.

Each is a small self-contained change. Methodology (benchmark-first, bit-identical, no-regression gate): see the Optimizing Scalar Expressions guide (PR #4933). Ranked by potential impact = per-row cost severity x usage frequency.

High impact (decimal aggregation hot path, common in TPC-DS)

• spark_unscaled_value (math_funcs/internal/unscaled_value.rs:37) - per-element Int64Builder loop casting Decimal128 -> Int64, rebuilding an identical null buffer. Fix: arr.unary::<_, Int64Type>(|v| v as i64). Injected by Spark's DecimalAggregates rule for sum/avg over narrow decimals.
• spark_make_decimal (math_funcs/internal/make_decimal.rs:45) - per-element Decimal128Builder loop, Int64 -> Decimal128. Fix: unary_opt (long_to_decimal already returns Option). Paired with UnscaledValue on the output side of the same rewrite.

Medium impact

• list_extract (array_funcs/list_extract.rs:285) - per-row MutableArrayData::extend(0, start+i, start+i+1) (one element/row) plus a Box<dyn Fn> index adjuster called per row. Fix: build a gather-index UInt32Array and issue one arrow::compute::take; monomorphize the index adjustment. Backs GetArrayItem (arr[i]) and element_at.
• cast_int_to_timestamp (conversion_funcs/numeric.rs:1006) - per-element TimestampMicrosecondBuilder loop over an infallible (v as i64) * MICROS_PER_SECOND. Fix: unary + with_timezone_opt.
• read_side_padding (rpad/CHAR(n)) (static_invoke/char_varchar_utils/read_side_padding.rs:213,275) - per-row chars().count() full scan (twice on the truncate branch) and char-by-char push for padding. Fix: scan length once; bulk-fill the pad run (extend/push_str of a cached pad slice). Runs on CHAR(n) columns.
• from_json / json_string_to_struct (json_funcs/from_json.rs:128) - allocates a full serde_json::Value tree per row just to read named fields, plus value.to_string() for non-string fields. Fix: arrow-json's streaming decoder (arrow_json::ReaderBuilder) or a borrowing visitor. Common in JSON ETL (not TPC-DS).
• NegativeExpr ANSI overflow check (math_funcs/negative.rs:62) - an extra full per-row scan for MIN before the (already vectorized) negation, in ANSI mode. Fix: eq_scalar(array, MIN) + any in one pass.
• spark_cast_decimal_to_boolean (conversion_funcs/numeric.rs:790) - per-element BooleanBuilder loop (!v.is_zero()). Fix: neq against a zero Decimal128 scalar, or single-pass iter().map().collect::<BooleanArray>().
• cast_float_to_timestamp (conversion_funcs/numeric.rs:1056) - per-element builder loop with NaN/inf/overflow checks. Fix: try_unary (ANSI) / unary_opt (Legacy/Try).
• cast_decimal_to_timestamp (conversion_funcs/numeric.rs:1030) - per-element builder loop; scale_factor already hoisted. Fix: unary (transform is infallible).

Low impact

• cast_boolean_to_timestamp (conversion_funcs/boolean.rs:60) - per-element builder loop (bool -> 0/1). Fix: single-pass iter().map().collect().
• cast_whole_num_to_binary (conversion_funcs/numeric.rs:213) - per-element BinaryBuilder loop; binary building is inherently element-wise, but the null branch can be dropped when null_count() == 0.
• date_trunc / timestamp_trunc array-format variants (kernels/temporal.rs:435,783) - per-row format.value(i).to_uppercase() allocation + per-row format->fn string match. Fix: eq_ignore_ascii_case (no allocation); resolve the trunc fn once per distinct format key. Only the array-of-formats path; the literal-format scalar path is already fast.
• array_position nested-type fallback (array_funcs/array_position.rs:264) - per-element ScalarValue::try_from_array allocation. Only triggered for nested/struct element types (primitives/strings/bools already have flat fast paths).
• json_array_length (json_funcs/json_array_length.rs:114) - near-optimal (streaming counter); minor once-per-scalar clone.

Notes

• Already optimized or in flight (excluded from the list): the decimal/int/float casts and narrowing int casts (PRs perf: optimize spark_array_compact (no-op fast path for null-free arrays) #4934, perf: optimize CheckOverflow with a shared no-overflow fast path (ANSI and non-ANSI) #4937-perf: vectorize float/decimal to narrow-integer casts #4941), plus the earlier campaign (spark_cast_int_to_int, to_json, parse_url, spark_size, spark_unhex, substring, date_trunc scalar path, spark_arrays_overlap, arrays_zip, get_json_object, regexp_extract/_all, spark_lpad, casts to string, etc.).
• Aggregate functions (agg_funcs/) were out of scope for this scan.
• Confirmed NOT candidates (already vectorized or inherently scalar): contains, split, map_sort, flatten, get_array_struct_fields, normalize_nan, spark_decimal_div, spark_modulo, WideDecimalBinaryExpr, murmur3/xxhash64 hashing, string-parse casts (utf8 -> int/float/decimal/timestamp), make_date/make_time, and rlike/rand (no Arrow kernel applies). All Regex::new calls are already hoisted into LazyLock statics.