perf: optimize CheckOverflow with a shared no-overflow fast path (ANSI and non-ANSI)

[andygrove]

Which issue does this PR close?

Part of #4936.

Rationale for this change

CheckOverflow wraps the result of every decimal + - * /, sum, and avg, so it runs on the hot path of most TPC-DS queries. Two allocations/scans were being paid even when no value overflowed, which is the common case:

• The non-ANSI path (Spark's default) always allocated a new Decimal128 array via null_if_overflow_precision.
• The ANSI path detected overflow with the array-level validate_decimal_precision, a non-inlined per-value function that also carries the error-formatting machinery. That made the ANSI no-overflow scan roughly 3x slower than the non-ANSI fast path even after buffer reuse.

What changes are included in this PR?

Both the ANSI and non-ANSI branches now share a single cheap fast-path scan built on is_valid_decimal_precision, a small inlined bounds check scanned with all (short-circuits at the first overflow). When no value overflows the target precision, both modes reuse the input buffers via to_data() (clones only cheap Arc metadata, no element copy).

The ANSI branch falls back to validate_decimal_precision only when an overflow is actually present, purely to build the precise Spark error for the first offending value. is_valid_decimal_precision and validate_decimal_precision compare identical precision bounds, so overflow detection is equivalent between the two. scale only affects error formatting, not the overflow decision.

Also adds array-path unit tests (only scalar inputs were covered before) and a criterion benchmark.

How are these changes tested?

New Rust unit tests cover the array path in both legacy and ANSI modes: no-overflow reuse (values, nulls, and target precision/scale preserved), overflow nulling in legacy mode, and overflow raising in ANSI mode. Existing scalar tests are unchanged.

Benchmark (criterion), baseline main vs this branch, 8192-row Decimal128 columns, two independent samples:

check_overflow: no overflow:              5.67 µs -> 5.19 µs   (-7 to -9%)
check_overflow: no overflow, with nulls:  7.40 µs -> 6.07 µs   (-18%)
check_overflow: ansi no overflow:        16.65 µs -> 5.13 µs   (-69%)
check_overflow: sparse overflow:          5.77 µs -> 5.75 µs   (within noise)
check_overflow: dense overflow:           8.78 µs -> 8.89 µs   (within noise)

The no-overflow shapes (the common case) hit the shared fast path. The ANSI no-overflow shape now reaches parity with the non-ANSI fast path instead of paying the heavier validate_decimal_precision scan. The overflow shapes run the unchanged null-masking path and are within noise across both samples.

[mbutrovich]

First pass, thanks @andygrove!

[mbutrovich]

On an ANSI batch that contains an overflow, the code now runs:

1. the fast-path all(is_valid_decimal_precision) scan (:128-131), which returns false,
2. validate_decimal_precision(*precision) (:140-141), a full second scan that finds the overflow and produces the error, and
3. inside the map_err, a third scan decimal_array.iter().find(...) (:146-158) to locate the first offending value for the Spark error.

Scans 2 and 3 are redundant. The fast-path scan at :128 already knows an overflow exists, and is_valid_decimal_precision already tells you which value is the first offender. You can drop validate_decimal_precision entirely and build the error from a single find:

} else if self.fail_on_error {
    // ANSI: fast path already proved an overflow exists. Find the first offending
    // value and raise the precise Spark error. Only runs on the aborting error path.
    let overflow_value = decimal_array
        .iter()
        .flatten()
        .find(|v| !Decimal128Type::is_valid_decimal_precision(*v, *precision))
        .unwrap_or(0);
    let spark_error =
        crate::error::decimal_overflow_error(overflow_value, *precision, *scale);
    return Err(match &self.query_context {
        Some(ctx) => DataFusionError::External(Box::new(
            crate::SparkErrorWithContext::with_context(spark_error, Arc::clone(ctx)),
        )),
        None => DataFusionError::External(Box::new(spark_error)),
    });
}

This removes the whole validate_decimal_precision call, the string-matching on "too large to store in a Decimal128" (which is brittle against arrow-rs wording changes, and also fails to catch the "too small" underflow branch at arrow-data/src/decimal.rs:1160), and the unreachable Internal error at :177-179. The PR says the error path "aborts the query anyway" so cost does not matter, but the simpler version is also more correct: the current find at :146-158 calls the three-arg Decimal128Type::validate_decimal_precision(val, precision, scale) and matches only the "too large" string, so a value that overflows on the negative side reaches the .unwrap_or(0) fallback and reports value 0 in the error. Using is_valid_decimal_precision in the find fixes that.

[andygrove]

Done in 65bfa33. Dropped validate_decimal_precision, the "too large" string-match, and the unreachable Internal error; the ANSI branch now builds the Spark error from a single find over is_valid_decimal_precision.

Confirmed the negative-overflow bug against arrow-data 58.3.0: validate_decimal_precision emits "too small to store" for a value below MIN_DECIMAL128_FOR_EACH_PRECISION, so the old outer match on "too large" fell through to DataFusionError::ArrowError and never produced the Spark error (and the inner find reported 0). is_valid_decimal_precision checks both bounds, so this is now correct for underflow too, covered by a new test.

[mbutrovich]

Every new array test uses a positive overflow value (1000 against precision 3). The MIN_DECIMAL128_FOR_EACH_PRECISION lower bound is never exercised. This matters because the existing ANSI error-formatting path only string-matches "too large" (see finding 1), so a negative overflow is a real untested branch. Add a legacy case that nulls a large-negative value and an ANSI case that errors on one:

#[test]
fn test_array_negative_overflow_nulled_legacy() {
    let batch = array_batch(vec![Some(-1000), Some(5)], 38, 0);
    let out = eval_array(&array_check_overflow(3, 0, false), &batch);
    assert_eq!(out.iter().collect::<Vec<_>>(), vec![None, Some(5)]);
}

[andygrove]

Added in 65bfa33: test_array_negative_overflow_nulled_legacy (nulls -1000 at precision 3, legacy) and test_array_negative_overflow_ansi_errors (raises on a negative overflow in ANSI). The second one guards the underflow branch that the old "too large" match missed.

[mbutrovich]

The review brief asks for all-overflow, all-null, and boundary-precision coverage. Current tests cover no-overflow, single-overflow, and mixed-null, but not:

• an all-null batch (the fast path takes flatten().all(...) which returns true on an empty iterator, so all-null must reuse the input and preserve the null mask - worth pinning),
• an all-overflow batch in legacy mode (every slot nulled) and ANSI mode (errors),
• a boundary value that exactly equals MAX_FOR_EACH_PRECISION[precision] (for example Some(999) at precision 3 is already the max and passes; add Some(9999) at precision 3 to confirm it is treated as overflow, since off-by-one on the bound is the classic failure).

These are cheap to add and directly guard the fast-path condition.

[andygrove]

Added in 65bfa33:

• test_array_all_null_reuses_input_and_preserves_mask (pins that flatten().all(...) returns true on all-null and the null mask survives),
• test_array_all_overflow_nulled_legacy and test_array_all_overflow_ansi_errors,
• test_array_boundary_precision_max_passes_and_over_by_one_overflows (999 passes, 9999 overflows at precision 3).

[mbutrovich]

#4937 detects overflow with all(is_valid_decimal_precision) over the raw input values. #4938 detects it with result.values().contains(&i128::MAX) over a sentinel that try_unary already wrote. The two do not and cannot share a helper, because in DecimalRescaleCheckOverflow the rescale pass has already folded the overflow decision into the sentinel, so re-deriving it from the original values would mean redoing the rescale. That is a defensible split, not a flaw. What is missing is the shared decision both PRs encode: "run the null-masking / error pass only when an overflow is actually present." If you want reuse, extract the non-ANSI tail both files share:

// returns the input array unchanged when nothing overflows, else the null-masked array
fn null_if_any_overflow(arr: &Decimal128Array, precision: u8, any_overflow: bool) -> Decimal128Array {
    if any_overflow { arr.null_if_overflow_precision(precision) } else { arr.clone() }
}

This is optional given the different detection inputs, but I would rather see one named concept than two ad hoc if shapes drifting apart over time. At minimum, land both PRs together so a reviewer can see the pattern is intentional.

[andygrove]

Leaving this one as-is. As you note, the detection inputs differ (this PR scans the raw input values; #4938 reads a post-rescale sentinel), so a shared helper would only wrap the tail if any_overflow { ... }, which is a single line in each file. I would rather not couple the two PRs around a one-line helper. Happy to land them together so the pattern is visible in review, and if #4938 goes in first I can rebase on top and extract the shared tail then if it still reads as one concept.

[parthchandra]

lgtm