diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 5cc07d04f2311..7302d6507e536 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -29,6 +29,7 @@ license: |
 - Since Spark 4.4, a nondeterministic common expression that a conditional branch reads more than once is evaluated once, and every read sees that one value. `BETWEEN` and `NULLIF` read their input twice, so an expression such as `CASE WHEN c THEN monotonically_increasing_id() BETWEEN 3 AND 5 END` previously gave each read its own value and could return a different result. Queries whose results depended on the earlier behaviour change, and an optimized or physical plan may now show a `with` node for such a branch. A definition that is cheap to evaluate and reports itself deterministic is still substituted at every read, as before.
 - Since Spark 4.4, for storage-partitioned joins, `spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be covered by some partition key instead of matching the partition keys positionally. As a result, a join-key column partitioned by more than one transform no longer prevents shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false` when the join keys are a subset of the partition keys. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`.
 - Since Spark 4.4, when `array_repeat` or `array_insert` is asked to build an array larger than the maximum supported array length, generated code raises the same error as interpreted evaluation. `array_repeat` now fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER` instead of the internal error `_LEGACY_ERROR_TEMP_2176`, and `array_insert` fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` instead of `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER`, which named a `count` parameter that `array_insert` does not have. Both functions raise an error under exactly the same conditions as before; only the reported error condition changes.
+- Since Spark 4.4, when `flatten` is asked to build an array larger than the maximum supported array length, generated code raises the same error as interpreted evaluation, `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION`, instead of the internal error `_LEGACY_ERROR_TEMP_2176`. `flatten` raises an error under exactly the same conditions as before; only the reported error condition changes.
 
 ## Upgrading from Spark SQL 4.2 to 4.3
 
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 3125753dc542f..2ecf53712520b 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
@@ -3448,11 +3448,17 @@ case class Flatten(child: Expression) extends UnaryExpression
       ctx: CodegenContext,
       childVariableName: String) : (String, String) = {
     val variableName = ctx.freshName("numElements")
+    // The upper bound is checked here rather than left to the array allocation so that this path
+    // reports the same error as `eval`. Without it the allocation fails with an internal error.
     val code = s"""
       |long $variableName = 0;
       |for (int z = 0; z < $childVariableName.numElements(); z++) {
       |  $variableName += $childVariableName.getArray(z).numElements();
       |}
+      |if ($variableName > ${ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH}) {
+      |  throw QueryExecutionErrors.arrayFunctionWithElementsExceedLimitError(
+      |    "$prettyName", $variableName);
+      |}
       """.stripMargin
     (code, variableName)
   }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
index a7cef98b4109a..5c868f31a234a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.analysis.{NamedParameter, UnresolvedGenerat
 import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Concat, CreateArray, EmptyRow, Expression, Flatten, Grouping, Literal, RowNumber, UnaryExpression, Years}
 import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode._
-import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode, GenerateUnsafeProjection}
 import org.apache.spark.sql.catalyst.expressions.objects.InitializeJavaBean
 import org.apache.spark.sql.catalyst.rules.RuleIdCollection
 import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLExpr
@@ -1357,18 +1357,27 @@ class QueryExecutionErrorsSuite
   test("Elements exceed limit for flatten()") {
     val array = new ColumnarArray(
       new ConstantColumnVector(Int.MaxValue, BooleanType), 0, Int.MaxValue)
+    val expr = Flatten(CreateArray(Seq(Literal.create(array, ArrayType(BooleanType)))))
 
-    checkError(
-      exception = intercept[SparkRuntimeException] {
-        Flatten(CreateArray(Seq(Literal.create(array, ArrayType(BooleanType))))).eval(EmptyRow)
-      },
-      condition = "COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION",
-      parameters = Map(
-        "numberOfElements" -> Int.MaxValue.toString,
-        "maxRoundedArrayLength" -> MAX_ROUNDED_ARRAY_LENGTH.toString,
-        "functionName" -> toSQLId("flatten")
+    // SPARK-59375: the generated code has to reject the length as well. Interpreted evaluation
+    // has always checked it, but codegen used to reach `ArrayData.allocateArrayData` and fail
+    // there with an internal error instead.
+    Seq(
+      () => expr.eval(EmptyRow),
+      () => GenerateUnsafeProjection.generate(Seq(expr)).apply(EmptyRow)
+    ).foreach { evaluate =>
+      checkError(
+        exception = intercept[SparkRuntimeException] {
+          evaluate()
+        },
+        condition = "COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION",
+        parameters = Map(
+          "numberOfElements" -> Int.MaxValue.toString,
+          "maxRoundedArrayLength" -> MAX_ROUNDED_ARRAY_LENGTH.toString,
+          "functionName" -> toSQLId("flatten")
+        )
       )
-    )
+    }
   }
 
   test("Elements exceed limit for array_repeat()") {