This is an automated email from the ASF dual-hosted git repository.

dtenedor pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.3 by this push:
     new 3d986d0edb6c [SPARK-59929][SQL] Fail fast when a window partition 
exceeds Int.MaxValue rows
3d986d0edb6c is described below

commit 3d986d0edb6ce8383c063b0f0b294cf19e6d4915
Author: Mark Jarvin <[email protected]>
AuthorDate: Mon Oct 5 10:23:25 2026 -0700

    [SPARK-59929][SQL] Fail fast when a window partition exceeds Int.MaxValue 
rows
    
    ### What changes were proposed in this pull request?
    
    `org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray` counted 
the buffered rows of a
    window partition in a 32-bit `Int` (`numRows`), exposed via `def length: 
Int`. When a single
    window partition exceeds `Int.MaxValue` (2,147,483,647) rows, the counter 
overflows and wraps
    negative, which silently corrupts the window-frame logic that reads it 
(most visibly, `lag`/`lead`
    returning `NULL` for every row; an `ArrayIndexOutOfBoundsException` is the 
sibling symptom at larger
    sizes).
    
    Window execution tracks the per-row index and window bounds in 32-bit 
`Int`s at several layers --
    the row counter above, the row index the driver (`WindowEvaluatorFactory` /
    `ArrowWindowPythonEvaluatorFactory`) passes to `WindowFunctionFrame.write`, 
the per-frame bounds,
    and the row accounting inside the spill-backed 
`UnsafeExternalSorter.getIterator`. Making window
    functions produce correct results over partitions larger than 
`Int.MaxValue` rows therefore
    requires widening all of these, which is a large change (notably the core 
sorter accounting, see
    the follow-up below).
    
    This PR takes the contained, safe step: it detects the condition and fails 
fast with a clear error
    rather than silently returning wrong results.
    
    - Widens the row counter and its accessor to `Long` (`numRows`/`length` in
      `ExternalAppendOnlyUnsafeRowArray`) so the true partition size can be 
observed instead of a
      wrapped negative value. `QueryExecutionErrors.invalidStartIndexError` is 
widened accordingly.
    - Adds a guard, `WindowEvaluatorFactoryBase.checkPartitionSizeLimit`, 
called by both window drivers
      after a partition is buffered. If the partition has more than 
`Int.MaxValue` rows it throws the
      new error condition `WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT` 
(SQLSTATE `54000`).
    - Uses `Math.toIntExact` at the remaining 32-bit narrowing sites in the 
frames and in
      `WindowSegmentTree` as a defensive backstop that fails loudly rather than 
truncating silently.
    
    Making the offset frames (`lag`/`lead`, unbounded-offset `nth_value`) 
actually produce correct
    results over such partitions requires widening the `UnsafeExternalSorter` 
row accounting to `Long`
    and is tracked as a follow-up:
    [SPARK-59941](https://issues.apache.org/jira/browse/SPARK-59941). Until 
then every window frame is
    guarded.
    
    This is the read-path counterpart of the write/insertion-path overflow 
fixed by
    [SPARK-51757](https://issues.apache.org/jira/browse/SPARK-51757), which did 
not widen
    `numRows`/`length`.
    
    ### Why are the changes needed?
    
    Window queries over a single partition larger than `Int.MaxValue` rows can 
return silently
    incorrect results. The clearest case, confirmed by reproduction, is 
`lag`/`lead` returning `NULL`
    for every row:
    
    ```sql
    SELECT count(*) AS n_rows, max(prev) AS max_prev, max(id) AS max_id
    FROM (
      SELECT id, lag(id, 1) OVER (PARTITION BY CAST(0 AS INT) ORDER BY id) AS 
prev
      FROM range(2200000000)
    );
    ```
    
    ```
    n_rows   = 2200000000     (all rows funnel to ONE partition via PARTITION 
BY a constant)
    max_prev = NULL           <-- BUG. Correct value is 2199999998
    max_id   = 2199999999     (sanity: the range itself is intact)
    ```
    
    `lag(id, 1)` over ids `0..2199999999` must yield `NULL, 0, 1, ..., 
2199999998`, so `max(prev)` must
    be `2199999998`; it returned `NULL` for every row. A query should either 
produce a correct result
    or fail with a clear error, never silently return wrong data.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes. A window query over a partition larger than `Int.MaxValue` rows now 
fails with the new error
    condition `WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT` (SQLSTATE `54000`) 
instead of silently
    returning wrong results (e.g. `lag`/`lead` returning `NULL` for every row) 
or throwing an
    `ArrayIndexOutOfBoundsException`. Partitions up to and including 
`Int.MaxValue` rows are unaffected.
    
    ### How was this patch tested?
    
    - New `WindowFunctionFrameSuite`: the partition-size guard allows 
`Int.MaxValue` rows and rejects
      anything larger with the expected error condition and message parameter.
    - New test in `ExternalAppendOnlyUnsafeRowArraySuite` pinning `length` as a 
`Long`.
    - Regression: `DataFrameWindowFunctionsSuite` and `WindowSegmentTreeSuite` 
pass unchanged;
      `SparkThrowableSuite` validates the new error condition.
    
    Materializing a 2^31+ row partition in a unit test is infeasible (the 
manual reproduction above
    took ~30 minutes over 2.2B rows on a dedicated core), so the overflow 
boundary is covered by the
    targeted unit tests above rather than an end-to-end run.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Isaac
    
    This pull request and its description were written by Isaac.
    
    Closes #59190 from markj-db/window-numrows-long-overflow.
    
    Authored-by: Mark Jarvin <[email protected]>
    Signed-off-by: Daniel Tenedorio <[email protected]>
    (cherry picked from commit 05afba131fdfae3866051036b2e3c1f8dc5b798b)
    Signed-off-by: Daniel Tenedorio <[email protected]>
---
 .../src/main/resources/error/error-conditions.json |  6 ++
 .../spark/sql/errors/QueryExecutionErrors.scala    | 10 +++-
 .../ExternalAppendOnlyUnsafeRowArray.scala         |  7 ++-
 .../python/ArrowWindowPythonEvaluatorFactory.scala |  3 +
 .../execution/window/WindowEvaluatorFactory.scala  |  3 +
 .../window/WindowEvaluatorFactoryBase.scala        | 16 ++++++
 .../sql/execution/window/WindowFunctionFrame.scala | 25 +++++++--
 .../sql/execution/window/WindowSegmentTree.scala   | 11 ++--
 .../ExternalAppendOnlyUnsafeRowArraySuite.scala    |  9 +++
 .../window/WindowFunctionFrameSuite.scala          | 64 ++++++++++++++++++++++
 10 files changed, 141 insertions(+), 13 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json 
b/common/utils/src/main/resources/error/error-conditions.json
index 05bc9b98dfa9..4e1a68a39790 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -9671,6 +9671,12 @@
     ],
     "sqlState" : "42601"
   },
+  "WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT" : {
+    "message" : [
+      "Cannot evaluate the window function because the window partition has 
<numRows> rows, which exceeds the maximum of 2,147,483,647 (2^31 - 1) rows 
supported per window partition. Reduce the partition size, for example by 
adding more expressions to the PARTITION BY clause."
+    ],
+    "sqlState" : "54000"
+  },
   "WINDOW_FUNCTION_WITHOUT_OVER_CLAUSE" : {
     "message" : [
       "Window function <funcName> requires an OVER clause."
diff --git 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 93c67612a18f..116e5cc0698c 100644
--- 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -2369,7 +2369,8 @@ private[sql] object QueryExecutionErrors extends 
QueryErrorsBase with ExecutionE
       cause = null)
   }
 
-  def invalidStartIndexError(numRows: Int, startIndex: Int): 
SparkArrayIndexOutOfBoundsException = {
+  def invalidStartIndexError(
+      numRows: Long, startIndex: Int): SparkArrayIndexOutOfBoundsException = {
     new SparkArrayIndexOutOfBoundsException(
       errorClass = "_LEGACY_ERROR_TEMP_2266",
       messageParameters = Map(
@@ -2387,6 +2388,13 @@ private[sql] object QueryExecutionErrors extends 
QueryErrorsBase with ExecutionE
         "className" -> className))
   }
 
+  def windowFunctionPartitionSizeExceedsLimitError(numRows: Long): 
SparkException = {
+    new SparkException(
+      errorClass = "WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT",
+      messageParameters = Map("numRows" -> numRows.toString()),
+      cause = null)
+  }
+
   def doExecuteBroadcastNotImplementedError(
       nodeName: String): SparkUnsupportedOperationException = {
     new SparkUnsupportedOperationException(
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
index 7ca9d9f85f48..c9cb12d68c14 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
@@ -87,7 +87,10 @@ class ExternalAppendOnlyUnsafeRowArray(
 
   private var spillableArray: UnsafeExternalSorter = _
   private var totalSpillBytes: Long = 0
-  private var numRows = 0
+  // The number of rows buffered. This is a `Long` because the spillable 
backing store can hold
+  // more than `Int.MaxValue` rows (e.g. a single large window partition); a 
32-bit counter would
+  // overflow and wrap negative, silently corrupting downstream length checks.
+  private var numRows: Long = 0
 
   // A counter to keep track of total modifications done to this array since 
its creation.
   // This helps to invalidate iterators when there are changes done to the 
backing array.
@@ -95,7 +98,7 @@ class ExternalAppendOnlyUnsafeRowArray(
 
   private var numFieldsPerRow = 0
 
-  def length: Int = numRows
+  def length: Long = numRows
 
   def isEmpty: Boolean = numRows == 0
 
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowWindowPythonEvaluatorFactory.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowWindowPythonEvaluatorFactory.scala
index ab9671c022f9..f6c0b8281acf 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowWindowPythonEvaluatorFactory.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowWindowPythonEvaluatorFactory.scala
@@ -319,6 +319,9 @@ class ArrowWindowPythonEvaluatorFactory(
             fetchNextRow()
           }
 
+          // Fail fast if the partition has more rows than window execution 
can handle.
+          checkPartitionSizeLimit(buffer.length)
+
           // Setup the frames.
           var i = 0
           while (i < numFrames) {
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
index 09c207ee603a..f4c986dc1a0b 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
@@ -115,6 +115,9 @@ class WindowEvaluatorFactory(
             fetchNextRow()
           }
 
+          // Fail fast if the partition has more rows than window execution 
can handle.
+          checkPartitionSizeLimit(buffer.length)
+
           // Setup the frames.
           var i = 0
           while (i < numFrames) {
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
index 40cba3d5ceb4..5ea56bc7133c 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
@@ -24,6 +24,7 @@ import org.apache.spark.{SparkException, TaskContext}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Add, 
AggregateWindowFunction, Ascending, Attribute, BoundReference, CurrentRow, 
DateAdd, DateAddYMInterval, DecimalAddNoOverflowCheck, Descending, Expression, 
ExtractANSIIntervalDays, FrameLessOffsetWindowFunction, FrameType, 
IdentityProjection, IntegerLiteral, MutableProjection, NamedExpression, 
OffsetWindowFunction, PythonFuncExpression, RangeFrame, RowFrame, RowOrdering, 
SortOrder, SpecifiedWindowFrame, TimestampAddInterval, TimestampA [...]
 import 
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, 
DeclarativeAggregate}
+import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{CalendarIntervalType, DateType, 
DayTimeIntervalType, DecimalType, IntegerType, TimestampNTZType, TimestampType, 
YearMonthIntervalType}
@@ -45,6 +46,21 @@ trait WindowEvaluatorFactoryBase {
   def numSegmentTreeFrames: Option[SQLMetric] = None
   def numSegmentTreeFallbackFrames: Option[SQLMetric] = None
 
+  /**
+   * Fail fast if the buffered partition has more rows than window execution 
can handle.
+   *
+   * Window execution tracks the row index within a partition and the window 
bounds in 32-bit
+   * `Int`s (including the row index the driver passes to 
[[WindowFunctionFrame.write]], and the
+   * row accounting in the spill-backed `UnsafeExternalSorter`). A partition 
with more than
+   * `Int.MaxValue` rows would overflow those counters and silently produce 
wrong results, so
+   * throw a clear user-facing error instead.
+   */
+  protected def checkPartitionSizeLimit(numRows: Long): Unit = {
+    if (numRows > Int.MaxValue) {
+      throw 
QueryExecutionErrors.windowFunctionPartitionSizeExceedsLimitError(numRows)
+    }
+  }
+
   /**
    * Create the resulting projection.
    *
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala
index 644603e4710f..e107bb067a43 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala
@@ -489,7 +489,10 @@ final class SlidingWindowFunctionFrame(
 
     // Only recalculate and update when the buffer changes.
     if (processor != null && bufferUpdated) {
-      processor.initialize(input.length)
+      // The window driver rejects partitions larger than Int.MaxValue rows 
before any frame runs
+      // (see WindowEvaluatorFactoryBase.checkPartitionSizeLimit); toIntExact 
is a defensive
+      // backstop that fails loudly rather than silently truncating if that 
guard is bypassed.
+      processor.initialize(Math.toIntExact(input.length))
       val iter = buffer.iterator()
       while (iter.hasNext) {
         processor.update(iter.next())
@@ -524,8 +527,12 @@ final class UnboundedWindowFunctionFrame(
 
   /** Prepare the frame for calculating a new partition. Process all rows 
eagerly. */
   override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
+    // The window driver rejects partitions larger than Int.MaxValue rows 
before any frame runs
+    // (see WindowEvaluatorFactoryBase.checkPartitionSizeLimit); toIntExact is 
a defensive
+    // backstop that fails loudly rather than silently truncating if that 
guard is bypassed.
+    val numRows = Math.toIntExact(rows.length)
     if (processor != null) {
-      processor.initialize(rows.length)
+      processor.initialize(numRows)
       val iterator = rows.generateIterator()
       while (iterator.hasNext) {
         processor.update(iterator.next())
@@ -534,7 +541,7 @@ final class UnboundedWindowFunctionFrame(
       processor.evaluate(target)
     }
 
-    upperBound = rows.length
+    upperBound = numRows
   }
 
   /** Write the frame columns for the current row to the given target row. */
@@ -595,7 +602,10 @@ final class UnboundedPrecedingWindowFunctionFrame(
     }
 
     if (processor != null) {
-      processor.initialize(input.length)
+      // The window driver rejects partitions larger than Int.MaxValue rows 
before any frame runs
+      // (see WindowEvaluatorFactoryBase.checkPartitionSizeLimit); toIntExact 
is a defensive
+      // backstop that fails loudly rather than silently truncating if that 
guard is bypassed.
+      processor.initialize(Math.toIntExact(input.length))
     }
   }
 
@@ -679,7 +689,10 @@ final class UnboundedFollowingWindowFunctionFrame(
 
     // Only recalculate and update when the buffer changes.
     if (processor != null && bufferUpdated) {
-      processor.initialize(input.length)
+      // The window driver rejects partitions larger than Int.MaxValue rows 
before any frame runs
+      // (see WindowEvaluatorFactoryBase.checkPartitionSizeLimit); toIntExact 
is a defensive
+      // backstop that fails loudly rather than silently truncating if that 
guard is bypassed.
+      processor.initialize(Math.toIntExact(input.length))
       if (nextRow != null) {
         processor.update(nextRow)
       }
@@ -692,5 +705,5 @@ final class UnboundedFollowingWindowFunctionFrame(
 
   override def currentLowerBound(): Int = inputIndex
 
-  override def currentUpperBound(): Int = input.length
+  override def currentUpperBound(): Int = Math.toIntExact(input.length)
 }
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
index cdc6556f1862..48f724b8f36b 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
@@ -237,12 +237,15 @@ private[window] class WindowSegmentTree(
    * Exception-safe: if aggregation throws, previously built state is 
preserved.
    */
   def build(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
-    // rows.length is Int by design; check guards against future widening.
-    val n = rows.length
-    if (n < 0) {
+    // The segment tree indexes rows with 32-bit Ints. The window driver 
rejects partitions
+    // larger than Int.MaxValue rows first (see 
WindowEvaluatorFactoryBase.checkPartitionSizeLimit);
+    // this check is a defensive backstop.
+    val numRowsLong = rows.length
+    if (numRowsLong > Int.MaxValue) {
       throw SparkException.internalError(
-        s"WindowSegmentTree cannot hold more than Int.MaxValue rows, got $n")
+        s"WindowSegmentTree cannot hold more than Int.MaxValue rows, got 
$numRowsLong")
     }
+    val n = numRowsLong.toInt
     val nBlocks = if (n == 0) 0 else (n + blockSize - 1) / blockSize
     val newBlockAggs = computeBlockAggregates(rows, n, nBlocks)
 
diff --git 
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArraySuite.scala
 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArraySuite.scala
index e667a95269f4..28a348b1a8af 100644
--- 
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArraySuite.scala
+++ 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArraySuite.scala
@@ -116,6 +116,15 @@ class ExternalAppendOnlyUnsafeRowArraySuite extends 
SparkFunSuite with LocalSpar
     assert(getNumBytesSpilled > 0)
   }
 
+  test("length is tracked as a Long") {
+    // `length` must return `Long` so a single partition can hold more than 
Int.MaxValue rows
+    // without the counter overflowing (window execution relies on this to 
detect oversized
+    // partitions). Assert the type directly: a plain value check would pass 
even if `length`
+    // reverted to `Int`, because small values widen implicitly.
+    
assert(classOf[ExternalAppendOnlyUnsafeRowArray].getMethod("length").getReturnType
 ===
+      java.lang.Long.TYPE)
+  }
+
   test("insert rows less than the inMemoryThreshold") {
     val (inMemoryThreshold, spillThreshold) = (100, 50)
     withExternalArray(inMemoryThreshold, spillThreshold) { array =>
diff --git 
a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowFunctionFrameSuite.scala
 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowFunctionFrameSuite.scala
new file mode 100644
index 000000000000..cc2cbcd04174
--- /dev/null
+++ 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowFunctionFrameSuite.scala
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.window
+
+import org.apache.spark.{SparkException, SparkFunSuite}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, 
NamedExpression, SortOrder}
+import org.apache.spark.sql.execution.metric.SQLMetric
+
+/**
+ * Tests for the window partition-size guard. Window execution tracks the 
per-row index and window
+ * bounds (and the spill-backed sorter's row accounting) in 32-bit `Int`s, so 
a partition larger
+ * than `Int.MaxValue` rows is rejected with a clear error instead of silently 
producing wrong
+ * results. See `WindowEvaluatorFactoryBase.checkPartitionSizeLimit`.
+ */
+class WindowFunctionFrameSuite extends SparkFunSuite {
+
+  /** Minimal factory exposing the protected partition-size guard for testing. 
*/
+  private class TestFactory extends WindowEvaluatorFactoryBase {
+    override def windowExpression: Seq[NamedExpression] = Nil
+    override def partitionSpec: Seq[Expression] = Nil
+    override def orderSpec: Seq[SortOrder] = Nil
+    override def childOutput: Seq[Attribute] = Nil
+    override def spillSize: SQLMetric = null
+
+    def check(numRows: Long): Unit = checkPartitionSizeLimit(numRows)
+  }
+
+  private val maxRows = Int.MaxValue.toLong
+
+  test("partitions up to Int.MaxValue rows are allowed") {
+    val factory = new TestFactory
+    factory.check(0L)
+    factory.check(maxRows)
+  }
+
+  test("partitions larger than Int.MaxValue rows are rejected with a clear 
error") {
+    val factory = new TestFactory
+    val e = intercept[SparkException] {
+      factory.check(maxRows + 1)
+    }
+    assert(e.getCondition == "WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT")
+    assert(e.getMessageParameters.get("numRows") == (maxRows + 1).toString)
+
+    // Far above the limit is rejected too.
+    intercept[SparkException] {
+      factory.check(Long.MaxValue)
+    }
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to