You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/04/23 07:23:06 UTC
(spark) branch master updated: [SPARK-56567][SQL] Fix array_insert `Int.MinValue` pos overflow
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 55ddfe30e7a0 [SPARK-56567][SQL] Fix array_insert `Int.MinValue` pos overflow
55ddfe30e7a0 is described below
commit 55ddfe30e7a086b2fd6f02f4ce1a7d2e80ad25de
Author: Akash Nayar <ak...@gmail.com>
AuthorDate: Thu Apr 23 15:22:49 2026 +0800
[SPARK-56567][SQL] Fix array_insert `Int.MinValue` pos overflow
### What changes were proposed in this pull request?
This PR fixes `array_insert` to raise the documented `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` error when called with `pos = Int.MinValue`, instead of leaking an internal JVM exception. Below is a minimalistic repro for the bug:
```
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ArrayInsertIntMinTest").getOrCreate()
def run(query):
print(query)
try:
spark.sql(query).show()
except Exception as e:
print(f"> {type(e).__name__}: {str(e).splitlines()[0]}")
print()
print("Correctly throws COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION")
run("SELECT array_insert(array(1), -2147483647, 3)")
print("Incorrectly throws ArrayIndexOutOfBoundsException")
run("SELECT array_insert(array(1), -2147483648, 3)")
spark.stop()
```
`array_insert` should raise `SparkRuntimeException` with condition `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` here. However, it instead crashes with an internal JVM exception: `java.lang.AssertionError` from `UnsafeArrayData.setInt(-2147483646, ...)` under whole-stage codegen, or `java.lang.ArrayIndexOutOfBoundsException: Index -2147483646 out of bounds for length 2` in interpreted mode. This is because `ArrayInsert` does `int` arithmetic on `pos` (e.g. `-pos`, `java.lang.Math.abs(pos) [...]
The fix is to widen `pos` arithmetic to `Long` in both `ArrayInsert.nullSafeEval` (interpreted) and `ArrayInsert.doGenCode` (codegen). The existing `MAX_ROUNDED_ARRAY_LENGTH` check then correctly catches the oversized result on the right branch.
### Why are the changes needed?
There is currently an error-reporting bug. `array_insert(arr, Int.MinValue, item)` fails with an internal JVM exception (`AssertionError` or `ArrayIndexOutOfBoundsException`) instead of the documented `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` error.
### Does this PR introduce _any_ user-facing change?
Yes, for `pos = Int.MinValue` with a non-empty input array, `array_insert` now throws `SparkRuntimeException` with condition `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` instead of leaking an internal `AssertionError` / `ArrayIndexOutOfBoundsException`. This matches the behavior already in place for neighboring positions like `Int.MinValue + 1`.
### How was this patch tested?
Added unit tests to `CollectionExpressionsSuite` to validate correct error reporting when `pos = Int.MinValue`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: claude-4.7-opus-high
Closes #55476 from aknayar/array-insert-pos-fix.
Authored-by: Akash Nayar <ak...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../expressions/collectionOperations.scala | 45 +++++++++++++---------
.../expressions/CollectionExpressionsSuite.scala | 18 +++++++++
2 files changed, 45 insertions(+), 18 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
index 113db0ba1ce3..60966f3098ca 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
@@ -5126,31 +5126,33 @@ case class ArrayInsert(
new GenericArrayData(newArray)
} else {
- var posInt = pos.asInstanceOf[Int]
- if (posInt == 0) {
+ // Widen `pos` to Long to avoid overflow (e.g. `-Int.MinValue` wraps back to `Int.MinValue`).
+ var posLong: Long = pos.asInstanceOf[Int].toLong
+ if (posLong == 0L) {
throw QueryExecutionErrors.invalidIndexOfZeroError(getContextOrNull())
}
- val newPosExtendsArrayLeft = (posInt < 0) && (-posInt > baseArr.numElements())
+ val newPosExtendsArrayLeft = (posLong < 0) && (-posLong > baseArr.numElements())
if (newPosExtendsArrayLeft) {
- val baseOffset = if (legacyNegativeIndex) 1 else 0
+ val baseOffset: Long = if (legacyNegativeIndex) 1L else 0L
// special case- if the new position is negative but larger than the current array size
// place the new item at start of array, place the current array contents at the end
// and fill the newly created array elements inbetween with a null
- val newArrayLength = -posInt + baseOffset
+ val newArrayLength: Long = -posLong + baseOffset
if (newArrayLength > ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH) {
throw QueryExecutionErrors.arrayFunctionWithElementsExceedLimitError(
prettyName, newArrayLength)
}
- val newArray = new Array[Any](newArrayLength)
+ val newArray = new Array[Any](newArrayLength.toInt)
baseArr.foreach(elementType, (i, v) => {
// current position, offset by new item + new null array elements
- val elementPosition = i + baseOffset + math.abs(posInt + baseArr.numElements())
+ val elementPosition =
+ (i + baseOffset + math.abs(posLong + baseArr.numElements())).toInt
newArray(elementPosition) = v
})
@@ -5158,20 +5160,21 @@ case class ArrayInsert(
new GenericArrayData(newArray)
} else {
- if (posInt < 0) {
- posInt = posInt + baseArr.numElements() + (if (legacyNegativeIndex) 0 else 1)
- } else if (posInt > 0) {
- posInt = posInt - 1
+ if (posLong < 0) {
+ posLong = posLong + baseArr.numElements() + (if (legacyNegativeIndex) 0 else 1)
+ } else if (posLong > 0) {
+ posLong = posLong - 1
}
- val newArrayLength = math.max(baseArr.numElements() + 1, posInt + 1)
+ val newArrayLength: Long = math.max(baseArr.numElements() + 1L, posLong + 1L)
if (newArrayLength > ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH) {
throw QueryExecutionErrors.arrayFunctionWithElementsExceedLimitError(
prettyName, newArrayLength)
}
- val newArray = new Array[Any](newArrayLength)
+ val newArray = new Array[Any](newArrayLength.toInt)
+ val posInt = posLong.toInt
baseArr.foreach(elementType, (i, v) => {
if (i >= posInt) {
@@ -5238,23 +5241,29 @@ case class ArrayInsert(
} else {
val pos = posExpr.value
val baseOffset = if (legacyNegativeIndex) 1 else 0
+ // Widen `pos` arithmetic to long so that `Int.MIN_VALUE` doesn't silently overflow
+ // (e.g. `Math.abs(Int.MIN_VALUE)` returns `Int.MIN_VALUE`).
+ val posLong = ctx.freshName("posLong")
+ val resLengthLong = ctx.freshName("resLengthLong")
s"""
+ |long $posLong = (long) $pos;
|int $itemInsertionIndex = 0;
|int $resLength = 0;
|int $adjustedAllocIdx = 0;
|boolean $insertedItemIsNull = ${itemExpr.isNull};
|
- |if ($pos == 0) {
+ |if ($posLong == 0L) {
| throw QueryExecutionErrors.invalidIndexOfZeroError($errorContext);
|}
|
- |if ($pos < 0 && (java.lang.Math.abs($pos) > $arr.numElements())) {
+ |if ($posLong < 0L && (-$posLong > $arr.numElements())) {
|
- | $resLength = java.lang.Math.abs($pos) + $baseOffset;
- | if ($resLength > ${ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH}) {
+ | long $resLengthLong = -$posLong + ${baseOffset}L;
+ | if ($resLengthLong > ${ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH}) {
| throw QueryExecutionErrors.arrayFunctionWithElementsExceedLimitError(
- | "$prettyName", $resLength);
+ | "$prettyName", $resLengthLong);
| }
+ | $resLength = (int) $resLengthLong;
|
| $allocation
| for (int $i = 0; $i < $arr.numElements(); $i ++) {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
index 10df53463793..e8da77e83433 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
@@ -3279,4 +3279,22 @@ class CollectionExpressionsSuite
a, Literal(5), Literal.create("q", StringType)), Seq("b", "a", "c", null, "q")
)
}
+
+ test("SPARK-56567: Array insert with pos = Int.MinValue") {
+ val a = Literal.create(Seq(1, 2, 4), ArrayType(IntegerType))
+ checkErrorInExpression[SparkRuntimeException](
+ ArrayInsert(a, Literal(Int.MinValue), Literal(3), legacyNegativeIndex = false),
+ condition = "COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION",
+ parameters = Map(
+ "numberOfElements" -> (-BigInt(Int.MinValue)).toString,
+ "maxRoundedArrayLength" -> ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH.toString,
+ "functionName" -> "`array_insert`"))
+ checkErrorInExpression[SparkRuntimeException](
+ ArrayInsert(a, Literal(Int.MinValue), Literal(3), legacyNegativeIndex = true),
+ condition = "COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION",
+ parameters = Map(
+ "numberOfElements" -> (-BigInt(Int.MinValue) + 1).toString,
+ "maxRoundedArrayLength" -> ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH.toString,
+ "functionName" -> "`array_insert`"))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org