dwsmith1983 opened a new pull request, #59159:
URL: https://github.com/apache/spark/pull/59159
### What changes were proposed in this pull request?
This supersedes #59076, which was closed before review. It is rebased on
current master and adds tests for the cast change, a negative-scale guard, and
aggregate test data that crosses the compact range in both directions.
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
