refactor: use arity helper for Int to Decimal128 reinterpretation

[0lai0]

Which issue does this PR close?

Part of #5091.

This is the first of a few small PRs splitting up #5091.
It covers the twoInt to Decimal128 reinterpretation sites
the remaining items (numeric.rs,array_insert.rs, temporal.rs, pow.rs, covariance.rs) will follow in separate PRs

Rationale for this change

Two sites hand-roll a per-row loop to turn an integer array into a Decimal128 array:

• native/core/src/execution/columnar_to_row.rs (maybe_cast_to_schema_type):
  Int32/Int64 to Decimal128, via iter().map().collect().
• native/spark-expr/src/math_funcs/internal/make_decimal.rs
  (spark_make_decimal): Int64 → Decimal128, via a Decimal128Builder loop.

What changes are included in this PR?

Replace the hand-rolled per-row loops with arity helpers:

• columnar_to_row.rs (maybe_cast_to_schema_type): Int32/Int64 → Decimal128 via arity::unary.
• make_decimal.rs (spark_make_decimal): Int64 → Decimal128 via unary_opt (non-ANSI, overflow → null) and try_unary (ANSI, overflow → error).

How are these changes tested?

• cargo test -p datafusion-comet-spark-expr
• cargo test -p datafusion-comet

[andygrove]

Thanks for splitting #5091 into reviewable pieces, and for being careful with the ANSI error path. I checked out the head commit and ran things rather than reading the diff alone. Recording what I verified so it doesn't have to be re-done:

• cargo test -p datafusion-comet-spark-expr make_decimal (5 passed) and cargo test -p datafusion-comet columnar_to_row (18 passed).
• cargo fmt --all -- --check and cargo clippy -p datafusion-comet-spark-expr -p datafusion-comet --all-targets -- -D warnings are both clean.
• The ANSI error is preserved exactly. The ArrowError::ExternalError unwrap does what the comment claims. A downcast test confirms the payload is still a SparkError, and the message is unchanged: External error: [NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION] 123456 cannot be represented as Decimal(3, 0).... This was the most likely thing to break silently and it doesn't.
• The first-offender ordering your new test asserts is real. try_unary walks valid indices through BitIndexIterator in ascending order and short-circuits on the first Err.
• Null handling is right. In arrow 58.4.0, unary_opt and try_unary both invoke the closure only on valid rows. unary in columnar_to_row.rs runs on every slot, but the widening is infallible so garbage under a null is harmless and the input null buffer carries over.
• Sliced inputs are fine. All three kernels go through values()/nulls(), which are already offset-adjusted. A scratch test on full.slice(2, 3) gives the right values and nulls.
• Dropping downcast_ref + ok_or_else in favour of as_primitive is safe. The match is on array.data_type(), so the old error arm was unreachable and this is not a new panic path.

So the risky parts went right. I have four things below, the first being the substantive one.

One process note: no CI checks have run on this branch yet, so a committer will need to approve the workflow run before this can merge.

[andygrove]

A question on the helper choice here. Arrow's own docs on try_unary note it "is often significantly slower than unary" because LLVM cannot vectorize fallible closures, and unary_opt has the same per-row shape. Meanwhile checkoverflow.rs, in this same directory and solving the same overflow problem, already uses a different idiom: widen unconditionally, then a short-circuiting is_valid_decimal_precision scan, paying for null-masking or error construction only when an overflow actually exists.

I benchmarked the three shapes on 8192-row Int64 batches at Decimal128(18, 2), the widest precision DecimalAggregates produces for MakeDecimal (times are non-ANSI / ANSI):

shape	no nulls	sparse nulls	dense nulls
builder loop (today)	19.0 / 19.1 µs	22.7 / 22.7 µs	19.1 / 19.2 µs
this PR	14.3 / 14.3 µs	15.8 / 14.9 µs	10.7 / 9.8 µs
unary + scan	5.58 / 8.50 µs	5.44 / 7.92 µs	5.48 / 9.04 µs

Your version is a genuine 1.3x to 1.9x win over what is there now. The checkoverflow.rs idiom looks like another 1.7x to 2.6x on top of that. MakeDecimal sits on the decimal sum and avg path, which is why its sibling got optimized in the first place, so it is worth getting right here.

Would you be up for trying that shape and seeing whether you reproduce the gain? The scan still finds the first offending value, so your new ordering test should keep passing. If you would rather stay with unary_opt/try_unary, could you note why in the PR description so the choice is on the record?

Either way, would you mind adding a make_decimal bench? benches/unscaled_value.rs and benches/check_overflow.rs cover the neighbouring expressions and would be easy to mirror. That would make the numbers checkable rather than something a reviewer has to measure by hand.

[andygrove]

PrimitiveArray::try_unary is generic over the error type. It is only the free function arity::try_unary that pins it to ArrowError. If you call the method form, as you already do for unary_opt just below, the closure can return DataFusionError directly:

arr.try_unary::<_, Decimal128Type, DataFusionError>(|v| {
    let v = v as i128;
    validate_decimal_precision(v, precision, scale)
        .map(|()| v)
        .map_err(|_| {
            DataFusionError::External(Box::new(decimal_overflow_error(v, precision, scale)))
        })
})?

I compiled and ran this and it produces the identical error, same variant and same SparkError payload. It drops the arrow::error::ArrowError import, the map_err match, and the other => arm that cannot currently be reached. It also removes the chance that a later edit loses the unwrap and turns the ANSI failure into a generic ArrowError, which would break the NUMERIC_VALUE_OUT_OF_RANGE mapping. This may become moot if you take the scan approach in my other comment.

[andygrove]

long_to_decimal is still here for the scalar path, and the array path now inlines the same validate-and-convert rule twice more. That is three copies with nothing keeping them in sync. Could one helper cover all three? Something like:

#[inline]
fn to_unscaled(v: i64, precision: u8, scale: i8) -> Option<i128> {
    let v = v as i128;
    validate_decimal_precision(v, precision, scale).ok().map(|()| v)
}

Then the non-ANSI array path becomes arr.unary_opt::<_, Decimal128Type>(|v| to_unscaled(v, precision, scale)), the ANSI path becomes to_unscaled(...).ok_or_else(|| overflow_err(v, precision, scale)), and long_to_decimal builds on the same two pieces.

[andygrove]

This reads well, and the note about an Arrow cast rescaling the value is worth having in the code. I confirmed that dropping the downcast_ref + ok_or_else is safe, since the match is on array.data_type(), so nothing new can panic here.

The substance of the change is null-buffer handling, and test_convert_int32_to_decimal128 and test_convert_int64_to_decimal128 both use all-valid arrays, so that is the one thing the tests do not exercise. Could you add a null to those arrays? A sliced input would be good too, following test_map_data_conversion_sliced_maparray further down the file. I checked both cases by hand and they behave correctly, so this is just about locking the behaviour in.