[SPARK-59805][SQL] Avoid BigDecimal round trips for wide decimals that fit in a long in UnsafeRow

[dwsmith1983]

What changes were proposed in this pull request?

For DECIMAL columns with precision > 18, UnsafeRow stores the unscaled value as big-endian two's-complement bytes. Reading and writing them always went through BigInteger/BigDecimal, even when the value fits in a long:

• UnsafeRow.getDecimal allocated a byte[], a BigInteger, a java.math.BigDecimal and a Scala BigDecimal per call, and returned a BigDecimal-backed Decimal.
• UnsafeRow.setDecimal and UnsafeRowWriter.write called toJavaBigDecimal().unscaledValue().toByteArray().

This PR adds a fast path for values whose unscaled value fits in a long:

1. UnsafeRow.getDecimal: when the stored value is at most 8 bytes, decode it directly from the row into a long. If |unscaled| < 10^18 and 0 <= scale <= 18, return a compact (long-backed) Decimal via Decimal.createUnsafe; otherwise return the same BigDecimal-backed Decimal as before (built with BigDecimal.valueOf, without the byte[]/BigInteger). Negative scales, possible only with spark.sql.legacy.allowNegativeScaleOfDecimal, keep the BigDecimal-backed result.
2. UnsafeRow.setDecimal / UnsafeRowWriter.write: after the existing changePrecision, a compact Decimal is written with the new UnsafeRow.writeCompactUnscaledBytes, which emits the minimal big-endian two's-complement bytes of the long. These are byte-identical to BigInteger.toByteArray(), so the row format, row hashing and row equality are unchanged. Non-compact values keep the existing path.
3. Decimal.roundToInt/roundToShort/roundToByte now truncate exactly via BigInteger for BigDecimal-backed values, as roundToLong already does, instead of going through a Double.

The main effect is on aggregation: the buffer of sum/avg over decimal(p, s) has precision p + 10, so e.g. sum(decimal(15,2)) uses a decimal(25,2) buffer. Previously every buffer read came back BigDecimal-backed, so Decimal.+ never took its existing long fast path; now it does while the running value fits in a long.

Why are the changes needed?

Performance. The per-row BigInteger/BigDecimal round trips are a large share of CPU in decimal-heavy aggregations. In a CPU profile of TPC-H SF10, 11.4% of samples were in java.math.BigDecimal/BigInteger or Decimal.

Item 3 is needed for consistency. With the double-based check, a value just below a bound, such as 2147483647.99999999 as decimal(20,8), rounds to 2147483648.0 and CAST(... AS INT) wrongly fails with CAST_OVERFLOW under ANSI mode. Long-backed values of the same number were already checked exactly. Because this PR returns more wide decimals in the compact form, leaving the double-based check would make the result of the same cast depend on the internal representation.

Does this PR introduce any user-facing change?

Yes, a bug fix. Casting a DECIMAL whose value truncates to exactly Int.MaxValue/Int.MinValue (and likewise for SMALLINT/TINYINT), but which is close enough to the next integer to round up past the bound as a double, no longer fails with CAST_OVERFLOW under ANSI mode (or returns null under try_cast). It returns the truncated value, as it already did for the same value held as an unscaled long. For example, CAST(CAST('2147483647.99999999' AS DECIMAL(20,8)) AS INT) now returns 2147483647 instead of failing.

No other change: query results, the UnsafeRow binary format and plans are unchanged.

How was this patch tested?

New tests:

• UnsafeRowWriterSuite: fixed-seed round trips for precision 19..38 across scales. Values include 0, +-1, +-(10^18 - 1), +-10^18, every byte-length boundary +-2^(8k-1) and its neighbours (including Long.MinValue/Long.MaxValue), values over 8 bytes, and random values. Each is written through UnsafeRowWriter and through in-place setDecimal. The tests assert:
  • the row bytes equal a layout built independently from BigInteger.toByteArray;
  • getDecimal matches the previous BigDecimal-based result in value, precision, scale, equals, hashCode and toString;
  • toLong, toInt, roundToLong, roundToInt, roundToShort, roundToByte, floor and ceil give the same value, or throw the same exception;
  • the result is compact exactly when |unscaled| < 10^18 and 0 <= scale <= 18.
• UnsafeRowWriterSuite also covers rescaling in the writer, overflow/null handling (null bit set, slot kept, re-update after null), and negative scales under spark.sql.legacy.allowNegativeScaleOfDecimal.
• CastWithAnsiOnSuite (inherited by TryCastSuite): casts of BigDecimal-backed decimal(20, s) values at the INT/SMALLINT/TINYINT bounds return the bound, and one unit in the last place beyond each bound fails with CAST_OVERFLOW (null under try_cast), interpreted and codegen.
• DecimalSuite: roundToInt/roundToShort/roundToByte on BigDecimal-backed values just inside and just outside each bound, compared with the long-backed representation.
• DataFrameAggregateSuite: sum/avg of decimal(15,2), grouped and global, with ANSI on and off. Within a single partition, running sums cross unscaled +-10^18 and come back below it, in both directions; other groups cross only when partial aggregates are merged.

Existing suites pass: DecimalSuite, UnsafeRowWriterSuite, UnsafeRowConverterSuite, DecimalExpressionSuite, CastWithAnsiOnSuite, CastWithAnsiOffSuite, TryCastSuite and DataFrameAggregateSuite. No golden files changed.

Benchmark: a new DecimalArithmeticBenchmark (5M rows, 1000 groups; ns per row; ANSI off / on):

Case	Before	After
sum(decimal(15,2))	83.3 / 80.4	48.2 / 45.4
avg(decimal(15,2))	79.9 / 78.9	47.0 / 47.4
sum(p * (1 - d) * (1 + t)), TPC-H q1 expression	165.9 / 168.6	157.6 / 159.2
sum(decimal(28,0)), all values >= 10^18	105.8 / 106.3	113.7 / 115.6 (within run-to-run noise of about +-5%)

The q1 expression doesn't improve because the multiply still produces BigDecimal-backed values; that is left for a follow-up. The *-results.txt files are not included; they should be generated with the GitHub Actions benchmark workflow.

End to end, TPC-H SF10 (local[4], median of 3 runs) takes 73.8s instead of 79.4s; q1 goes from 11.55s to 8.76s. TPC-DS SF10 is unchanged within noise (its decimals are decimal(7,2), whose sums fit in 18 digits).

Was this patch authored or co-authored using generative AI tooling?

No