markj-db opened a new pull request, #58014:
URL: https://github.com/apache/spark/pull/58014
### What changes were proposed in this pull request?
This hardens the shuffle order-independent (row-based) checksum
(`spark.sql.shuffle.orderIndependentChecksum.enabled`, on by default)
against a class of JVM
crashes.
`UnsafeRowChecksum` computes each row's checksum by reading the row's raw
backing memory via
`XXH64.hashUnsafeBytes` -> `Platform.getLong`, with no validation. When a
row's `UnsafeRow` carries
an invalid pointer -- a null `baseObject` with a near-zero `baseOffset`, or
a negative size -- that
unchecked read dereferences bad memory and the JVM crashes with a SIGSEGV
(`si_addr=0x0`). Because
the checksum is enabled by default, a single latent bad-row bug surfaces as
a process crash that
takes the whole executor down.
This PR adds a validation step that runs in `RowBasedChecksum.update`
*before* the row is ever
dereferenced:
- `RowBasedChecksum` gains an overridable `validateRow` hook, a row-ordinal
counter, and an
overridable `failOnInvalidRow` member (default true). A flagged row is
logged (via the
structured-logging framework) with full context (stage / partition / task
/ row-ordinal and the
row's `baseObject`, `baseOffset`, `sizeInBytes`) and then, by default,
fails the task with a
descriptive error.
- `UnsafeRowChecksum` implements the check for `UnsafeRow` backing memory
and overrides
`failOnInvalidRow` via its constructor. It flags only what a
validly-constructed `UnsafeRow` can
never be -- a negative size, a null off-heap `baseObject` pointing into
the first memory page (the
`si_addr=0x0` shape), or an out-of-bounds `byte[]` offset. Any other
non-null on-heap base (e.g.
the common `long[]` buffer) is accepted, and a corrupt-but-large size is
deliberately not flagged,
so a well-formed row is never flagged.
- A new internal conf
`spark.sql.shuffle.orderIndependentChecksum.failOnInvalidRow` (default `true`)
exists only as a safety valve: set it to `false` to log and disable that
partition's checksum
instead of failing, should a validator false positive ever be observed.
`ShuffleExchangeExec`
reads it when it constructs the per-partition checksums.
Failing is strictly better than the status quo: today a bad row crashes the
JVM with a SIGSEGV,
which fails the task anyway (via retry exhaustion) and is more disruptive
because it takes the whole
executor down. Failing the task with a clear error never increases query
failures for a genuinely
bad row and makes it debuggable. The invalid pointer is never dereferenced
in either mode. A stale
off-heap pointer at a *plausible* (high) address cannot be distinguished
from a live one here and is
not caught -- that would need the separate "checksum the serialized bytes"
change.
### Why are the changes needed?
The order-independent shuffle checksum is enabled by default, so any bad
`UnsafeRow` reaching it
crashes the executor with a SIGSEGV instead of failing the task with an
actionable error. Guarding
the read turns an opaque process crash into a loggable, attributable task
failure without changing
behavior for well-formed rows.
### Does this PR introduce _any_ user-facing change?
No. For well-formed rows behavior is unchanged. A genuinely invalid row that
would previously have
crashed the JVM now fails the task with a descriptive `SparkException` (or,
with the new
safety-valve conf disabled, is logged and the partition's checksum is
skipped).
### How was this patch tested?
Unit tests added to the existing `UnsafeRowChecksumSuite` (sql/core):
- `validate` accepts well-formed on-heap rows (both `byte[]`- and
`long[]`-backed) and empty rows,
and flags a null-`baseObject` first-page pointer (the `si_addr=0x0` shape)
and a negative size; a
large-but-positive size is deliberately not flagged.
- Fail mode (the default, including the no-arg constructor) raises a
descriptive `SparkException` on
an invalid row.
- Recover mode (`failOnInvalidRow = false`) disables the checksum on an
invalid row without throwing
or dereferencing, and stays disabled for subsequent rows.
- `createUnsafeRowChecksums` threads the `failOnInvalidRow` flag through to
the checksum objects.
Ran locally (JDK 17, SBT): `UnsafeRowChecksumSuite` (9 tests) and
`LogKeysSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Yes -- authored using Claude Code (Opus 4.8), reviewed by the author.
--
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]
