dtenedor commented on code in PR #58014:
URL: https://github.com/apache/spark/pull/58014#discussion_r3808519599
##########
core/src/main/java/org/apache/spark/shuffle/checksum/RowBasedChecksum.scala:
##########
@@ -19,19 +19,35 @@ package org.apache.spark.shuffle.checksum
import scala.util.control.NonFatal
-import org.apache.spark.internal.Logging
+import org.apache.spark.{SparkException, TaskContext}
+import org.apache.spark.internal.{Logging, LogKeys}
/**
* A class for computing checksum for input (key, value) pairs. The checksum
is independent of
* the order of the input (key, value) pairs. It is done by computing a
checksum for each row
* first, then computing the XOR and SUM for all the row checksums and mixing
these two values
* as the final checksum.
+ *
+ * [[failOnInvalidRow]] controls what happens when [[validateRow]] flags a row
whose backing
+ * memory is unsafe to read: true (default) logs the row's context and fails
the task with a
+ * descriptive error; false logs and disables this checksum so the query
proceeds (a safety valve
+ * for a validator false positive). In neither case is the invalid pointer
dereferenced, so a
+ * corrupt row no longer crashes the JVM with a SIGSEGV.
Review Comment:
> In neither case is the invalid pointer dereferenced, so a corrupt row no
longer crashes the JVM with a SIGSEGV
The first clause is right, but the conclusion doesn't follow for
`failOnInvalidRow = false`. In the `ExternalSorter` path the row is already
sitting in the buffer when update runs (`ExternalSorter.insertAll`, the
`buffer.insert`/`maybeSpillCollection`/`update` sequence), and it later gets
`Platform.copyMemory`'d by `UnsafeRowSerializer.writeValue` to
`UnsafeRow.writeToStream` at spill/write time.
So with a genuinely bad row in recover mode the executor still dies, just
further downstream and now detached from the diagnostic log you just wrote.
Suggest scoping the sentence to the checksum: "...so the checksum itself never
faults on a corrupt pointer" and adding that recover mode is intended only as a
false-positive escape hatch, not as a way to survive a real bad row.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRowChecksum.scala:
##########
@@ -47,7 +55,55 @@ class UnsafeRowChecksum extends RowBasedChecksum() {
}
object UnsafeRowChecksum {
- def createUnsafeRowChecksums(numPartitions: Int): Array[RowBasedChecksum] = {
- Array.tabulate(numPartitions)(_ => new UnsafeRowChecksum())
+ // Off-heap rows carry an absolute address in baseOffset. Real allocations
sit far above the
+ // first page; an address inside it is the null-region read behind the
SIGSEGV (si_addr=0x0).
+ private val MinNativeAddress: Long = 4096L
+
+ /**
+ * Checks that `row`'s backing memory can be safely hashed. Returns
Some(description) only for a
+ * row that a validly-constructed UnsafeRow can never be -- a negative size,
a null baseObject
+ * (off-heap) pointing into the first memory page, or an out-of-bounds
byte[] offset -- so a
+ * well-formed row is never flagged. Any other non-null on-heap base (e.g. a
long[]) is accepted:
+ * it is a live Java array, so a read within it cannot fault at a null
address. These are the
+ * cases that make XXH64's unchecked reads fault. Note: a stale off-heap
pointer to a *freed* page
+ * at a plausible (high) address cannot be distinguished from a live one
here and is not caught;
+ * nor is a corrupt-but-large size, a heuristic we deliberately avoid to
keep false positives
+ * impossible.
+ */
+ def validate(row: UnsafeRow): Option[String] = {
Review Comment:
This becomes a public Catalyst API if the method is public; is this
intended? Else we can make it `private[spark]`.
##########
core/src/main/java/org/apache/spark/shuffle/checksum/RowBasedChecksum.scala:
##########
@@ -50,21 +66,62 @@ abstract class RowBasedChecksum() extends Serializable with
Logging {
/** Updates the row-based checksum with the given (key, value) pair. Not
thread safe. */
def update(key: Any, value: Any): Unit = {
if (!hasError) {
- try {
- val rowChecksumValue = calculateRowChecksum(key, value)
- checksumXor = checksumXor ^ rowChecksumValue
- checksumSum += rowChecksumValue
- } catch {
- case NonFatal(e) =>
- logError("Checksum computation encountered error: ", e)
- hasError = true
+ rowOrdinal += 1
+ // Guard the row before calculateRowChecksum touches its memory: a
checksum that reads raw
+ // row bytes would otherwise SIGSEGV on a corrupt pointer instead of
reporting it.
+ validateRow(value) match {
+ case Some(problem) => reportInvalidRow(problem)
+ case None =>
+ try {
+ val rowChecksumValue = calculateRowChecksum(key, value)
+ checksumXor = checksumXor ^ rowChecksumValue
+ checksumSum += rowChecksumValue
+ } catch {
+ case NonFatal(e) =>
+ logError(log"Checksum computation encountered error", e)
+ hasError = true
+ }
}
}
}
/** Computes and returns the checksum value for the given (key, value) pair
*/
protected def calculateRowChecksum(key: Any, value: Any): Long
+ /**
+ * Validates that `value`'s backing memory is safe to read, called before
[[calculateRowChecksum]]
+ * dereferences it. Returns Some(description) if the row must not be read
(the description is
+ * logged and, in fail mode, becomes the error message), or None if it looks
well-formed. The
+ * default accepts every row; subclasses that read raw row memory override
this.
+ */
+ protected def validateRow(value: Any): Option[String] = None
+
+ // Handles a row flagged by validateRow. Always logs the full context; then
either fails the task
+ // (failOnInvalidRow) or disables this checksum and lets the query proceed.
Runs before the row is
+ // dereferenced, turning what would have been a JVM crash into a loggable,
attributable event.
+ private def reportInvalidRow(problem: String): Unit = {
+ val tc = TaskContext.get()
+ val stage = if (tc != null) tc.stageId() else -1
+ val partition = if (tc != null) tc.partitionId() else -1
+ val taskAttempt = if (tc != null) tc.taskAttemptId() else -1L
+ val context =
+ log"${MDC(LogKeys.REASON, problem)} (stage=${MDC(LogKeys.STAGE_ID,
stage)} " +
+ log"partition=${MDC(LogKeys.PARTITION_ID, partition)} " +
+ log"taskAttempt=${MDC(LogKeys.TASK_ATTEMPT_ID, taskAttempt)} " +
+ log"rowOrdinal=${MDC(LogKeys.ROW_INDEX, rowOrdinal)})"
+ if (failOnInvalidRow) {
+ logError(log"Invalid row in shuffle row-based checksum: " + context)
+ throw SparkException.internalError(
Review Comment:
We set `hasError = true` below on L121, but if we ever catch this exception
later, it could be possible for `getValue` to return a checksum computed over a
strict subset of the partition's rows. That sounds bad/possibly incorrect;
should we just set `hasError = true` above on L111 instead?
##########
sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowChecksumSuite.scala:
##########
@@ -146,4 +147,73 @@ class UnsafeRowChecksumSuite extends SparkFunSuite {
assert(rowBasedChecksum1.getValue != 0)
assert(rowBasedChecksum2.getValue != 0)
}
+
+ // --- Invalid-row guard ---
+ // Production crashed with a SIGSEGV in XXH64.hashBytesByWords ->
Platform.getLong (si_addr=0x0)
+ // when the checksum read an UnsafeRow whose backing pointer was invalid.
The guard flags such a
+ // row before it is dereferenced and either recovers (default) or fails the
task.
+
+ // The production crash shape: an off-heap row (null baseObject) whose
baseOffset points into the
+ // first memory page. Reading it would fault near si_addr 0x0.
+ private def nullLowAddressRow(size: Int = 16): UnsafeRow = {
+ val row = new UnsafeRow(1)
+ row.pointTo(null, 0L, size)
+ row
+ }
+
+ test("validate accepts a well-formed row and flags invalid backing memory") {
+
assert(UnsafeRowChecksum.validate(toUnsafeRow(Row(20)).asInstanceOf[UnsafeRow]).isEmpty)
+
+ // Empty row: nothing is dereferenced.
+ val empty = new UnsafeRow(1)
+ empty.pointTo(null, 0L, 0)
+ assert(UnsafeRowChecksum.validate(empty).isEmpty)
+
+ // A long[]-backed on-heap row (a common UnsafeRow buffer) must be
accepted, not mistaken for
+ // an invalid base type.
+ val longBacked = new UnsafeRow(1)
+ longBacked.pointTo(new Array[Long](2), Platform.LONG_ARRAY_OFFSET, 16)
+ assert(UnsafeRowChecksum.validate(longBacked).isEmpty)
+
+ // Null baseObject pointing into the first page (the si_addr=0x0 crash).
+
assert(UnsafeRowChecksum.validate(nullLowAddressRow()).exists(_.contains("first
memory page")))
+
+ // Negative size (corrupt size field). A large-but-positive size is
deliberately NOT flagged.
+ val negSize = new UnsafeRow(1)
+ negSize.pointTo(null, 0x100000L, -8)
+ assert(UnsafeRowChecksum.validate(negSize).exists(_.contains("negative
sizeInBytes")))
+ }
+
+ test("recover mode disables the checksum on an invalid row instead of
crashing") {
+ val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = false)
+ // Must neither throw nor dereference the bad pointer; getValue then
returns the default 0.
+ rowBasedChecksum.update(0, nullLowAddressRow())
+ assert(rowBasedChecksum.getValue == 0L)
+ // A subsequent valid row does not revive the checksum once it is in the
error state.
+ rowBasedChecksum.update(0, toUnsafeRow(Row(20)))
+ assert(rowBasedChecksum.getValue == 0L)
+ }
+
+ test("fail mode (the default) raises a descriptive error on an invalid row")
{
+ val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = true)
+ val e = intercept[SparkException] {
+ rowBasedChecksum.update(0, nullLowAddressRow())
+ }
+ assert(e.getMessage.contains("Invalid row in shuffle row-based checksum"))
+ // The no-arg constructor defaults to fail mode.
+ intercept[SparkException] {
+ new UnsafeRowChecksum().update(0, nullLowAddressRow())
+ }
+ }
+
+ test("createUnsafeRowChecksums threads the fail-on-invalid-row flag") {
+ val recover = UnsafeRowChecksum.createUnsafeRowChecksums(1,
failOnInvalidRow = false)
+ recover(0).update(0, nullLowAddressRow())
+ assert(recover(0).getValue == 0L)
+
+ val fail = UnsafeRowChecksum.createUnsafeRowChecksums(1, failOnInvalidRow
= true)
+ intercept[SparkException] {
+ fail(0).update(0, nullLowAddressRow())
+ }
Review Comment:
Other test case ideas:
No coverage for the `byte[]` out-of-bounds branch:
`UnsafeRowChecksum.validate`'s case bytes: `Array[Byte]` branch
(`UnsafeRowChecksum.scala`:92-96) has no test at all. Because `pointTo` rejects
that shape outright, writing the test forces you through `setTotalSize`:
```scala
val oob = new UnsafeRow(1)
oob.pointTo(new Array[Byte](8), Platform.BYTE_ARRAY_OFFSET, 8)
oob.setTotalSize(64) // the only way past pointTo's own bounds check
assert(UnsafeRowChecksum.validate(oob).exists(_.contains("on-heap row
spans")))
```
The accept side of the off-heap check is untested, including the boundary.
Nothing asserts validate accepts a null-`baseObject` row at a real address,
which is the false-positive risk that matters most for a fail-by-default guard.
And nothing pins `MinNativeAddress`:
```scala
val addr = Platform.allocateMemory(16)
try {
val offheap = new UnsafeRow(1)
offheap.pointTo(null, addr, 16)
assert(UnsafeRowChecksum.validate(offheap).isEmpty)
} finally Platform.freeMemory(addr)
// Boundary: 4095 is flagged, 4096 is not.
val atBoundary = new UnsafeRow(1)
atBoundary.pointTo(null, 4096L, 8)
assert(UnsafeRowChecksum.validate(atBoundary).isEmpty)
```
No end-to-end coverage that the conf reaches the checksum: every test here
constructs the checksum directly, so nothing exercises
`spark.sql.shuffle.orderIndependentChecksum.failOnInvalidRow` through
`ShuffleExchangeExec.scala`:574-575/582-583. If that plumbing regressed, this
suite stays green. `MapStatusEndToEndSuite` (`sql/core`) already drives the
sibling `orderIndependentChecksum.enabled` conf end to end and looks like the
natural home; at minimum a test that the conf's default is true and that a
shuffle with well-formed rows is unaffected in both modes.
##########
sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowChecksumSuite.scala:
##########
@@ -146,4 +147,73 @@ class UnsafeRowChecksumSuite extends SparkFunSuite {
assert(rowBasedChecksum1.getValue != 0)
assert(rowBasedChecksum2.getValue != 0)
}
+
+ // --- Invalid-row guard ---
+ // Production crashed with a SIGSEGV in XXH64.hashBytesByWords ->
Platform.getLong (si_addr=0x0)
+ // when the checksum read an UnsafeRow whose backing pointer was invalid.
The guard flags such a
+ // row before it is dereferenced and either recovers (default) or fails the
task.
+
+ // The production crash shape: an off-heap row (null baseObject) whose
baseOffset points into the
+ // first memory page. Reading it would fault near si_addr 0x0.
+ private def nullLowAddressRow(size: Int = 16): UnsafeRow = {
+ val row = new UnsafeRow(1)
+ row.pointTo(null, 0L, size)
+ row
+ }
+
+ test("validate accepts a well-formed row and flags invalid backing memory") {
+
assert(UnsafeRowChecksum.validate(toUnsafeRow(Row(20)).asInstanceOf[UnsafeRow]).isEmpty)
+
+ // Empty row: nothing is dereferenced.
+ val empty = new UnsafeRow(1)
+ empty.pointTo(null, 0L, 0)
+ assert(UnsafeRowChecksum.validate(empty).isEmpty)
+
+ // A long[]-backed on-heap row (a common UnsafeRow buffer) must be
accepted, not mistaken for
+ // an invalid base type.
+ val longBacked = new UnsafeRow(1)
+ longBacked.pointTo(new Array[Long](2), Platform.LONG_ARRAY_OFFSET, 16)
+ assert(UnsafeRowChecksum.validate(longBacked).isEmpty)
+
+ // Null baseObject pointing into the first page (the si_addr=0x0 crash).
+
assert(UnsafeRowChecksum.validate(nullLowAddressRow()).exists(_.contains("first
memory page")))
+
+ // Negative size (corrupt size field). A large-but-positive size is
deliberately NOT flagged.
+ val negSize = new UnsafeRow(1)
+ negSize.pointTo(null, 0x100000L, -8)
+ assert(UnsafeRowChecksum.validate(negSize).exists(_.contains("negative
sizeInBytes")))
+ }
+
+ test("recover mode disables the checksum on an invalid row instead of
crashing") {
+ val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = false)
+ // Must neither throw nor dereference the bad pointer; getValue then
returns the default 0.
+ rowBasedChecksum.update(0, nullLowAddressRow())
+ assert(rowBasedChecksum.getValue == 0L)
+ // A subsequent valid row does not revive the checksum once it is in the
error state.
+ rowBasedChecksum.update(0, toUnsafeRow(Row(20)))
+ assert(rowBasedChecksum.getValue == 0L)
Review Comment:
These assertions can't currently fail: `getValue == 0L` is also what a
freshly constructed checksum returns, so this test passes even if update does
nothing at all. We could establish a non-zero baseline first:
```scala
test("recover mode disables the checksum on an invalid row instead of
crashing") {
val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = false)
rowBasedChecksum.update(0, toUnsafeRow(Row(20)))
assert(rowBasedChecksum.getValue != 0L) // meaningful starting state
// Must neither throw nor dereference the bad pointer.
rowBasedChecksum.update(0, nullLowAddressRow())
assert(rowBasedChecksum.getValue == 0L) // the bad row disabled it
rowBasedChecksum.update(0, toUnsafeRow(Row(40)))
assert(rowBasedChecksum.getValue == 0L) // and it stays disabled
}
```
##########
core/src/main/java/org/apache/spark/shuffle/checksum/RowBasedChecksum.scala:
##########
@@ -19,19 +19,35 @@ package org.apache.spark.shuffle.checksum
import scala.util.control.NonFatal
-import org.apache.spark.internal.Logging
+import org.apache.spark.{SparkException, TaskContext}
+import org.apache.spark.internal.{Logging, LogKeys}
/**
* A class for computing checksum for input (key, value) pairs. The checksum
is independent of
* the order of the input (key, value) pairs. It is done by computing a
checksum for each row
* first, then computing the XOR and SUM for all the row checksums and mixing
these two values
* as the final checksum.
+ *
+ * [[failOnInvalidRow]] controls what happens when [[validateRow]] flags a row
whose backing
+ * memory is unsafe to read: true (default) logs the row's context and fails
the task with a
+ * descriptive error; false logs and disables this checksum so the query
proceeds (a safety valve
+ * for a validator false positive). In neither case is the invalid pointer
dereferenced, so a
+ * corrupt row no longer crashes the JVM with a SIGSEGV.
*/
abstract class RowBasedChecksum() extends Serializable with Logging {
private val ROTATE_POSITIONS = 27
private var hasError: Boolean = false
private var checksumXor: Long = 0
private var checksumSum: Long = 0
+ // Rows passed to `update` so far (1-based for the current row); reported
with an invalid row to
+ // locate it within its partition. Not part of the checksum.
+ private var rowOrdinal: Long = 0
+
+ /**
+ * When true (the default), a row flagged by [[validateRow]] fails the task;
when false, it only
+ * disables this checksum. Overridden by concrete subclasses (typically from
a constructor arg).
+ */
+ protected def failOnInvalidRow: Boolean = true
Review Comment:
Is this the right default that we want?
`RowBasedChecksum` currently documents itself as best-effort: line 53-54:
"returns the default checksum value (0) if there are any errors encountered
during the checksum computation", and every other failure mode (a non-UnsafeRow
value, any exception out of the hash) disables the checksum rather than failing
the query. Making this one error class fail the task inverts that contract for
one case.
The alternative would be to throw from
`UnsafeRowChecksum.calculateRowChecksum` and let the existing catch `NonFatal`
on lines 79-83 disable the checksum. No `validateRow` hook, no
`reportInvalidRow`, no new conf, no new `LogKey`, and one less public method in
catalyst. All of the new machinery exists purely to make the default "fail"
instead of "disable".
##########
sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowChecksumSuite.scala:
##########
@@ -146,4 +147,73 @@ class UnsafeRowChecksumSuite extends SparkFunSuite {
assert(rowBasedChecksum1.getValue != 0)
assert(rowBasedChecksum2.getValue != 0)
}
+
+ // --- Invalid-row guard ---
+ // Production crashed with a SIGSEGV in XXH64.hashBytesByWords ->
Platform.getLong (si_addr=0x0)
+ // when the checksum read an UnsafeRow whose backing pointer was invalid.
The guard flags such a
+ // row before it is dereferenced and either recovers (default) or fails the
task.
+
+ // The production crash shape: an off-heap row (null baseObject) whose
baseOffset points into the
+ // first memory page. Reading it would fault near si_addr 0x0.
+ private def nullLowAddressRow(size: Int = 16): UnsafeRow = {
Review Comment:
We never call this with a non-default `size`, let's either remove the param
or else call it with a non-default value?
##########
sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowChecksumSuite.scala:
##########
@@ -146,4 +147,73 @@ class UnsafeRowChecksumSuite extends SparkFunSuite {
assert(rowBasedChecksum1.getValue != 0)
assert(rowBasedChecksum2.getValue != 0)
}
+
+ // --- Invalid-row guard ---
+ // Production crashed with a SIGSEGV in XXH64.hashBytesByWords ->
Platform.getLong (si_addr=0x0)
+ // when the checksum read an UnsafeRow whose backing pointer was invalid.
The guard flags such a
+ // row before it is dereferenced and either recovers (default) or fails the
task.
+
+ // The production crash shape: an off-heap row (null baseObject) whose
baseOffset points into the
+ // first memory page. Reading it would fault near si_addr 0x0.
+ private def nullLowAddressRow(size: Int = 16): UnsafeRow = {
+ val row = new UnsafeRow(1)
+ row.pointTo(null, 0L, size)
+ row
+ }
+
+ test("validate accepts a well-formed row and flags invalid backing memory") {
+
assert(UnsafeRowChecksum.validate(toUnsafeRow(Row(20)).asInstanceOf[UnsafeRow]).isEmpty)
+
+ // Empty row: nothing is dereferenced.
+ val empty = new UnsafeRow(1)
+ empty.pointTo(null, 0L, 0)
+ assert(UnsafeRowChecksum.validate(empty).isEmpty)
+
+ // A long[]-backed on-heap row (a common UnsafeRow buffer) must be
accepted, not mistaken for
+ // an invalid base type.
+ val longBacked = new UnsafeRow(1)
+ longBacked.pointTo(new Array[Long](2), Platform.LONG_ARRAY_OFFSET, 16)
+ assert(UnsafeRowChecksum.validate(longBacked).isEmpty)
+
+ // Null baseObject pointing into the first page (the si_addr=0x0 crash).
+
assert(UnsafeRowChecksum.validate(nullLowAddressRow()).exists(_.contains("first
memory page")))
+
+ // Negative size (corrupt size field). A large-but-positive size is
deliberately NOT flagged.
+ val negSize = new UnsafeRow(1)
+ negSize.pointTo(null, 0x100000L, -8)
+ assert(UnsafeRowChecksum.validate(negSize).exists(_.contains("negative
sizeInBytes")))
+ }
+
+ test("recover mode disables the checksum on an invalid row instead of
crashing") {
+ val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = false)
+ // Must neither throw nor dereference the bad pointer; getValue then
returns the default 0.
+ rowBasedChecksum.update(0, nullLowAddressRow())
+ assert(rowBasedChecksum.getValue == 0L)
+ // A subsequent valid row does not revive the checksum once it is in the
error state.
+ rowBasedChecksum.update(0, toUnsafeRow(Row(20)))
+ assert(rowBasedChecksum.getValue == 0L)
+ }
+
+ test("fail mode (the default) raises a descriptive error on an invalid row")
{
+ val rowBasedChecksum = new UnsafeRowChecksum(failOnInvalidRow = true)
+ val e = intercept[SparkException] {
+ rowBasedChecksum.update(0, nullLowAddressRow())
+ }
+ assert(e.getMessage.contains("Invalid row in shuffle row-based checksum"))
Review Comment:
Can we use `checkError` for this?
##########
sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRowChecksum.scala:
##########
@@ -47,7 +55,55 @@ class UnsafeRowChecksum extends RowBasedChecksum() {
}
object UnsafeRowChecksum {
- def createUnsafeRowChecksums(numPartitions: Int): Array[RowBasedChecksum] = {
- Array.tabulate(numPartitions)(_ => new UnsafeRowChecksum())
+ // Off-heap rows carry an absolute address in baseOffset. Real allocations
sit far above the
+ // first page; an address inside it is the null-region read behind the
SIGSEGV (si_addr=0x0).
+ private val MinNativeAddress: Long = 4096L
+
+ /**
+ * Checks that `row`'s backing memory can be safely hashed. Returns
Some(description) only for a
+ * row that a validly-constructed UnsafeRow can never be -- a negative size,
a null baseObject
+ * (off-heap) pointing into the first memory page, or an out-of-bounds
byte[] offset -- so a
+ * well-formed row is never flagged. Any other non-null on-heap base (e.g. a
long[]) is accepted:
+ * it is a live Java array, so a read within it cannot fault at a null
address. These are the
+ * cases that make XXH64's unchecked reads fault. Note: a stale off-heap
pointer to a *freed* page
+ * at a plausible (high) address cannot be distinguished from a live one
here and is not caught;
+ * nor is a corrupt-but-large size, a heuristic we deliberately avoid to
keep false positives
+ * impossible.
+ */
+ def validate(row: UnsafeRow): Option[String] = {
+ val base = row.getBaseObject
+ val offset = row.getBaseOffset
+ val size = row.getSizeInBytes
+ def desc(reason: String): Option[String] = {
+ val baseStr = if (base == null) "null" else base.getClass.getName
+ Some(s"$reason (baseObject=$baseStr,
baseOffset=0x${java.lang.Long.toHexString(offset)}, " +
+ s"sizeInBytes=$size)")
+ }
+ if (size < 0) {
+ desc("negative sizeInBytes")
Review Comment:
A negative sizeInBytes is not a crash vector, so this one is a genuine new
failure mode.
The PR's central argument is "failing never increases query failures,
because the row would have SIGSEGV'd anyway." That doesn't hold here:
`XXH64.hashUnsafeBytes` computes `end = offset + length`, which is below
`offset` when `length` is negative, so `hashBytesByWords` skips its length >=
32 block and every subsequent `while (offset <= limit) / while (offset < end)`
loop is false; nothing is read. The assert `(length >= 0)` on `XXH64.java:93`
fires under `-ea` in tests but is off in production executors.
So today a negative-size row yields a garbage-but-harmless hash; after this
PR it fails the task. Flagging it is defensible (a negative size is corruption)
but worth considering whether this behavior as a replacement for a crash is
intended.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRowChecksum.scala:
##########
@@ -47,7 +55,55 @@ class UnsafeRowChecksum extends RowBasedChecksum() {
}
object UnsafeRowChecksum {
- def createUnsafeRowChecksums(numPartitions: Int): Array[RowBasedChecksum] = {
- Array.tabulate(numPartitions)(_ => new UnsafeRowChecksum())
+ // Off-heap rows carry an absolute address in baseOffset. Real allocations
sit far above the
+ // first page; an address inside it is the null-region read behind the
SIGSEGV (si_addr=0x0).
+ private val MinNativeAddress: Long = 4096L
+
+ /**
+ * Checks that `row`'s backing memory can be safely hashed. Returns
Some(description) only for a
+ * row that a validly-constructed UnsafeRow can never be -- a negative size,
a null baseObject
+ * (off-heap) pointing into the first memory page, or an out-of-bounds
byte[] offset -- so a
+ * well-formed row is never flagged. Any other non-null on-heap base (e.g. a
long[]) is accepted:
+ * it is a live Java array, so a read within it cannot fault at a null
address. These are the
+ * cases that make XXH64's unchecked reads fault. Note: a stale off-heap
pointer to a *freed* page
+ * at a plausible (high) address cannot be distinguished from a live one
here and is not caught;
+ * nor is a corrupt-but-large size, a heuristic we deliberately avoid to
keep false positives
+ * impossible.
+ */
+ def validate(row: UnsafeRow): Option[String] = {
+ val base = row.getBaseObject
+ val offset = row.getBaseOffset
+ val size = row.getSizeInBytes
+ def desc(reason: String): Option[String] = {
+ val baseStr = if (base == null) "null" else base.getClass.getName
+ Some(s"$reason (baseObject=$baseStr,
baseOffset=0x${java.lang.Long.toHexString(offset)}, " +
+ s"sizeInBytes=$size)")
+ }
+ if (size < 0) {
+ desc("negative sizeInBytes")
+ } else if (size == 0) {
+ None // empty row: XXH64 reads nothing
+ } else {
+ base match {
+ case null =>
+ if (offset < MinNativeAddress) {
+ desc("off-heap row (null baseObject) points into the first memory
page")
+ } else None
+ case bytes: Array[Byte] =>
+ val start = offset - Platform.BYTE_ARRAY_OFFSET
+ if (start < 0 || start + size > bytes.length) {
+ desc(s"on-heap row spans [$start, ${start + size}) of a
${bytes.length}-byte array")
+ } else None
Review Comment:
This `byte[]` bounds check duplicates a check `pointTo` already enforces:
```scala
if (baseObject instanceof byte[] bytes) {
int offsetInByteArray = (int) (baseOffset - Platform.BYTE_ARRAY_OFFSET);
if (offsetInByteArray < 0 || sizeInBytes < 0 ||
bytes.length < offsetInByteArray + sizeInBytes) {
throw new SparkIllegalArgumentException(...);
```
So the only way to reach lines 94-96 is the `setTotalSize` back door, which
mutates `sizeInBytes` without re-validating. As written the branch reads as if
it catches something `pointTo` lets through. Either drop it, or keep it and say
in the comment that it covers the `setTotalSize` window, and add the test for
it which is what will settle whether it's reachable at all.
--
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]
