[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)`), which silently wraps back to `Int.MinValue`. The `newPosExtendsArrayLeft` check therefore returns the wrong answer, the code takes the wrong branch, never reaches the `MAX_ROUNDED_ARRAY_LENGTH` guard, and computes a garbage index that crashes later.

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 apache#55476 from aknayar/array-insert-pos-fix.

Authored-by: Akash Nayar <akashknayar5@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: aknayar

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -5126,52 +5126,55 @@ 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
         })
 
         newArray(0) = item
 
         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 ++) {

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`"))
+  }
 }