This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new d2a108c19d46 [SPARK-59824][SQL] Guard top-level StringType/BinaryType
inputs to pickle Python UDFs
d2a108c19d46 is described below
commit d2a108c19d4609b00901f1bca909216c73169e23
Author: Tengfei Huang <[email protected]>
AuthorDate: Fri Oct 9 08:42:42 2026 +0000
[SPARK-59824][SQL] Guard top-level StringType/BinaryType inputs to pickle
Python UDFs
### What changes were proposed in this pull request?
**Problem**
For non-Arrow Python UDFs, Spark evaluates the UDF input expressions and
serializes the projected row using Java conversion followed by pickle
serialization.
A single row can contain a very large string or binary value. Because
batching cannot split an individual row, converting and pickling it may create
several large in-memory copies and cause an executor OOM before the row reaches
the Python worker.
Existing batch-size limits do not protect against this case.
**Proposed change**
Add an optional row-size guard for pickle-based Python UDFs.
The guard should:
- Run after UDF input expressions are projected, so computed values such as
cast, concat, and substring are measured.
- Run before Java conversion and pickle serialization.
- Estimate the payload size of supported top-level variable-width
arguments, initially strings and binary values.
- Fail with a structured error when the estimated size exceeds a
configurable threshold based on executor heap.
- Introduce no per-row overhead when disabled or when the projected schema
has no supported variable-width arguments.
Pls note that: Nested arrays, maps, and structs are intentionally excluded
initially to avoid a recursive traversal of every input value.
### Why are the changes needed?
Avoid single giant row crashing executors.
### Does this PR introduce _any_ user-facing change?
Yes.
Previously, a single oversized input row for a pickle-serialized Python UDF
was converted and serialized without a row-level size check, potentially
causing executor OOM or a Python worker crash.
This PR adds an opt-in guard that checks the combined payload size of
top-level string and binary arguments after projection and before conversion
and pickling. If the configured executor-heap fraction is exceeded, the query
fails with
UDF_LIMITS.ROW_SIZE.
The guard is disabled by default, so default behavior remains unchanged.
This capability is new relative to both released Spark versions and the current
master branch.
### How was this patch tested?
UTs
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: ClaudeCode Opus 4.8
Closes #59122 from ivoson/SPARK-59824.
Authored-by: Tengfei Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit d1e511b6387fcca474d4a9beefc332114dfd2707)
Signed-off-by: Yicong-Huang <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 13 ++
.../spark/sql/errors/QueryExecutionErrors.scala | 9 ++
.../org/apache/spark/sql/internal/SQLConf.scala | 27 ++++
.../sql/execution/python/BatchEvalPythonExec.scala | 94 +++++++++++-
.../python/BatchEvalPythonExecSuite.scala | 167 ++++++++++++++++++++-
5 files changed, 300 insertions(+), 10 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index c4ecebab14a0..eed97372cfe3 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8374,6 +8374,19 @@
],
"sqlState" : "54011"
},
+ "UDF_LIMITS" : {
+ "message" : [
+ "A UDF resource limit was exceeded."
+ ],
+ "subClass" : {
+ "ROW_SIZE" : {
+ "message" : [
+ "The projected input row size <actualRowSize> bytes exceeds the
maximum allowed size <maxRowSize> bytes."
+ ]
+ }
+ },
+ "sqlState" : "54000"
+ },
"UDTF_ALIAS_NUMBER_MISMATCH" : {
"message" : [
"The number of aliases supplied in the AS clause does not match the
number of columns output by the UDTF.",
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 f80cfecf7aa8..66662e9dc6bb 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
@@ -198,6 +198,15 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
cause = e)
}
+ def pythonUDFRowSizeExceededError(
+ maxRowSize: Long, actualRowSize: Long): RuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "UDF_LIMITS.ROW_SIZE",
+ messageParameters = Map(
+ "maxRowSize" -> maxRowSize.toString,
+ "actualRowSize" -> actualRowSize.toString))
+ }
+
def divideByZeroError(context: QueryContext): ArithmeticException = {
new SparkArithmeticException(
errorClass = "DIVIDE_BY_ZERO",
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index de2f850f9d14..dff899c005ae 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -5485,6 +5485,33 @@ object SQLConf {
"be -1 (no limit) or greater than zero and less than or equal to
INT_MAX.")
.createWithDefault(-1)
+ val PYTHON_UDF_ROW_SIZE_GUARD_ENABLED =
+ buildConf("spark.sql.execution.python.udf.rowSizeGuard.enabled")
+ .internal()
+ .doc("When true, guard pickle-serialized (non-Arrow) Python UDF
evaluation against " +
+ "top-level string and binary argument payloads whose combined size
exceeds " +
+ "rowSizeGuard.maxRowHeapFraction of executor heap. The guard checks
the projected " +
+ "arguments before conversion and pickling. Nested inputs are not
estimated.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
+ val PYTHON_UDF_ROW_SIZE_GUARD_MAX_ROW_HEAP_FRACTION =
+ buildConf("spark.sql.execution.python.udf.rowSizeGuard.maxRowHeapFraction")
+ .internal()
+ .doc("Maximum combined byte size of one projected input row's top-level
string and " +
+ "binary Python UDF arguments, as a fraction of executor max heap. The
default 0.083 " +
+ "is approximately 1/12 of the executor heap; conversion and pickling
can require " +
+ "multiple copies of the argument bytes.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doubleConf
+ .checkValue(v => v > 0.0 && v <= 1.0,
+ "The value of
spark.sql.execution.python.udf.rowSizeGuard.maxRowHeapFraction " +
+ "must be in (0.0, 1.0].")
+ .createWithDefault(0.083)
+
val PYTHON_UDF_BUFFER_SIZE =
buildConf("spark.sql.execution.python.udf.buffer.size")
.doc(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
index 4c39ce98cf55..3707543ee57e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
@@ -27,11 +27,12 @@ import org.apache.spark.{JobArtifactSet, TaskContext}
import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.execution.python.EvalPythonExec.ArgumentMetadata
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.types.{BinaryType, DataType, StringType,
StructField, StructType}
/**
* A physical plan that evaluates a [[PythonUDF]]
@@ -60,7 +61,10 @@ case class BatchEvalPythonExec(udfs: Seq[PythonUDF],
resultAttrs: Seq[Attribute]
pythonMetrics,
jobArtifactUUID,
sessionUUID,
- binaryAsBytes)
+ binaryAsBytes,
+ rowSizeGuardEnabled =
conf.getConf(SQLConf.PYTHON_UDF_ROW_SIZE_GUARD_ENABLED),
+ rowSizeGuardMaxRowHeapFraction =
+ conf.getConf(SQLConf.PYTHON_UDF_ROW_SIZE_GUARD_MAX_ROW_HEAP_FRACTION))
}
override protected def withNewChildInternal(newChild: SparkPlan):
BatchEvalPythonExec =
@@ -76,9 +80,43 @@ class BatchEvalPythonEvaluatorFactory(
pythonMetrics: Map[String, SQLMetric],
jobArtifactUUID: Option[String],
sessionUUID: Option[String],
- binaryAsBytes: Boolean)
+ binaryAsBytes: Boolean,
+ rowSizeGuardEnabled: Boolean = false,
+ rowSizeGuardMaxRowHeapFraction: Double = -1.0)
extends EvalPythonEvaluatorFactory(childOutput, udfs, output) {
+ // This lazy val is first used in a task, even when the factory is built on
the driver.
+ private lazy val rowSizeGuardMaxRowBytes = if
(rowSizeGuardMaxRowHeapFraction > 0) {
+ (Runtime.getRuntime.maxMemory() * rowSizeGuardMaxRowHeapFraction).toLong
+ } else {
+ -1L
+ }
+
+ protected def projectedInputCheck(
+ inputTypes: Seq[DataType]): Option[InternalRow => Unit] = {
+ if (rowSizeGuardEnabled && rowSizeGuardMaxRowBytes >= 0) {
+ BatchEvalPythonExec.projectedInputPayloadSizeEstimator(inputTypes).map {
estimate =>
+ row => {
+ val estimatedBytes = estimate(row)
+ if (estimatedBytes > rowSizeGuardMaxRowBytes) {
+ throw QueryExecutionErrors.pythonUDFRowSizeExceededError(
+ rowSizeGuardMaxRowBytes, estimatedBytes)
+ }
+ }
+ }
+ } else {
+ None
+ }
+ }
+
+ private[python] def getInputIterator(
+ iter: Iterator[InternalRow],
+ schema: StructType): Iterator[Array[Byte]] = {
+ BatchEvalPythonExec.getInputIterator(
+ iter, schema, batchSize, binaryAsBytes, maxBytesPerBatch, pythonMetrics,
+ projectedInputCheck(schema.map(_.dataType)))
+ }
+
override def evaluate(
funcs: Seq[(ChainedPythonFunctions, Long)],
argMetas: Array[Array[ArgumentMetadata]],
@@ -88,8 +126,7 @@ class BatchEvalPythonEvaluatorFactory(
EvaluatePython.registerPicklers() // register pickler for Row
// Input iterator to Python.
- val inputIterator = BatchEvalPythonExec.getInputIterator(
- iter, schema, batchSize, binaryAsBytes, maxBytesPerBatch, pythonMetrics)
+ val inputIterator = getInputIterator(iter, schema)
// Output iterator for results from Python.
val outputIterator =
@@ -125,13 +162,48 @@ class BatchEvalPythonEvaluatorFactory(
}
object BatchEvalPythonExec {
+ /**
+ * Estimates top-level variable-width payloads in a projected UDF argument
row. Nested inputs are
+ * intentionally excluded to avoid a recursive walk of every value.
+ */
+ private[python] def projectedInputPayloadSizeEstimator(
+ inputTypes: Seq[DataType]): Option[InternalRow => Long] = {
+ val projectedPayloads = inputTypes.zipWithIndex.flatMap { case (dataType,
ordinal) =>
+ dataType match {
+ case _: StringType | BinaryType => Some((ordinal, dataType))
+ case _ => None
+ }
+ }.toArray
+ if (projectedPayloads.isEmpty) {
+ None
+ } else {
+ Some { row =>
+ var estimatedBytes = 0L
+ var i = 0
+ while (i < projectedPayloads.length) {
+ val (ordinal, dataType) = projectedPayloads(i)
+ if (!row.isNullAt(ordinal)) {
+ estimatedBytes += (dataType match {
+ case _: StringType => row.getUTF8String(ordinal).numBytes.toLong
+ case BinaryType => row.getBinary(ordinal).length.toLong
+ case _ => 0L
+ })
+ }
+ i += 1
+ }
+ estimatedBytes
+ }
+ }
+ }
+
def getInputIterator(
iter: Iterator[InternalRow],
schema: StructType,
batchSize: Int,
binaryAsBytes: Boolean,
maxBytesPerBatch: Long = -1L,
- pythonMetrics: Map[String, SQLMetric] = Map.empty):
Iterator[Array[Byte]] = {
+ pythonMetrics: Map[String, SQLMetric] = Map.empty,
+ projectedInputCheck: Option[InternalRow => Unit] = None):
Iterator[Array[Byte]] = {
val peakPickledBatchBytesMetric =
pythonMetrics.get("pythonPeakPickledBatchBytes")
val oversizedBatchMetric = pythonMetrics.get("pythonOversizedBatchCount")
val estimatedInputBytesMetric =
pythonMetrics.get("pythonEstimatedInputBytes")
@@ -155,7 +227,7 @@ object BatchEvalPythonExec {
// Converts a row to the java object pickled to Python. When `sizeAcc` is
defined, toJava also
// accumulates the per-row pickled-size estimate at its leaf cases during
this same traversal
// (no second walk).
- def convertRow(row: InternalRow, sizeAcc: Option[PickledSizeAccumulator]):
Any = {
+ def convertRowUnchecked(row: InternalRow, sizeAcc:
Option[PickledSizeAccumulator]): Any = {
if (needConversion) {
EvaluatePython.toJava(row, schema, binaryAsBytes, sizeAcc)
} else {
@@ -171,6 +243,14 @@ object BatchEvalPythonExec {
fields
}
}
+ val convertRow: (InternalRow, Option[PickledSizeAccumulator]) => Any =
+ projectedInputCheck match {
+ case Some(check) => (row, sizeAcc) => {
+ check(row)
+ convertRowUnchecked(row, sizeAcc)
+ }
+ case None => convertRowUnchecked
+ }
// Input iterator to Python: input rows are grouped so we send them in
batches to Python.
val batchedIter: Iterator[Array[Any]] =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExecSuite.scala
index 34d63df710cb..74b80cfc3202 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExecSuite.scala
@@ -20,13 +20,19 @@ package org.apache.spark.sql.execution.python
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
-import org.apache.spark.api.python.{PythonEvalType, SimplePythonFunction}
+import org.apache.spark.{SparkRuntimeException, TaskContext}
+import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType,
SimplePythonFunction}
import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference,
GenericInternalRow, GreaterThan, In}
+import org.apache.spark.sql.catalyst.expressions.{
+ And, AttributeReference, Concat, GenericInternalRow, GreaterThan, In,
PythonUDF,
+ UnsafeProjection}
import org.apache.spark.sql.connector.catalog.CatalogManager
-import org.apache.spark.sql.execution.{FilterExec, InputAdapter,
WholeStageCodegenExec}
+import org.apache.spark.sql.execution.{
+ FilterExec, InputAdapter, LocalTableScanExec, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.execution.python.EvalPythonExec.ArgumentMetadata
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.{ExamplePointUDT, SharedSparkSession}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -81,6 +87,161 @@ class BatchEvalPythonExecSuite extends SharedSparkSession
assert(metric.value === expectedPeak)
}
+ test("SPARK-59824: projected UDF input payload estimator covers evaluated
values") {
+ val inputTypes = Seq(StringType, BinaryType, IntegerType)
+ val projectedRow = InternalRow(
+ UTF8String.fromString("s" * 2000), Array.fill[Byte](4000)(1), 1)
+
+ val estimate = BatchEvalPythonExec
+ .projectedInputPayloadSizeEstimator(inputTypes)
+ .get
+ assert(estimate(projectedRow) === 6000L)
+ assert(estimate(InternalRow(null, null, 1)) === 0L)
+
+ // Fixed-width and nested inputs are intentionally not estimated.
+ assert(BatchEvalPythonExec
+ .projectedInputPayloadSizeEstimator(Seq(IntegerType,
ArrayType(StringType)))
+ .isEmpty)
+ }
+
+ test("SPARK-59824: pickle UDF row-size guard rejects a computed projected
argument") {
+ // Exercise the production evaluator wiring without starting a Python
worker.
+ val errors = spark.sparkContext.parallelize(Seq(1), 1).mapPartitions { _ =>
+ val taskAttr = AttributeReference("s", StringType)()
+ val taskRow = UnsafeProjection.create(Array[DataType](StringType))
+ .apply(InternalRow(UTF8String.fromString("x" * 600)))
+ val projectedArg = Concat(Seq(taskAttr, taskAttr))
+ val pythonUdf = PythonUDF(
+ "dummy", new DummyUDF, StringType, Seq(projectedArg),
PythonEvalType.SQL_BATCHED_UDF,
+ udfDeterministic = true)
+ val maxRowBytes = 1000L
+ val maxRowHeapFraction = maxRowBytes.toDouble /
Runtime.getRuntime.maxMemory()
+ val expectedMaxRowSize =
+ (Runtime.getRuntime.maxMemory() * maxRowHeapFraction).toLong
+ val factory = new BatchEvalPythonEvaluatorFactory(
+ childOutput = Seq(taskAttr),
+ udfs = Seq(pythonUdf),
+ output = Seq.empty,
+ batchSize = 1,
+ maxBytesPerBatch = -1L,
+ pythonMetrics = Map.empty,
+ jobArtifactUUID = None,
+ sessionUUID = None,
+ binaryAsBytes = false,
+ rowSizeGuardEnabled = true,
+ rowSizeGuardMaxRowHeapFraction = maxRowHeapFraction) {
+ override def evaluate(
+ funcs: Seq[(ChainedPythonFunctions, Long)],
+ argMetas: Array[Array[ArgumentMetadata]],
+ iter: Iterator[InternalRow],
+ schema: StructType,
+ context: TaskContext): Iterator[InternalRow] = {
+ getInputIterator(iter, schema).next()
+ Iterator.empty
+ }
+ }
+ try {
+ factory.createEvaluator().eval(0, Iterator(taskRow)).hasNext
+ Iterator.empty[(SparkRuntimeException, Long)]
+ } catch {
+ case e: SparkRuntimeException => Iterator((e, expectedMaxRowSize))
+ }
+ }.collect().toSeq
+ assert(errors.length === 1)
+ val (exception, expectedMaxRowSize) = errors.head
+ checkError(
+ exception = exception,
+ condition = "UDF_LIMITS.ROW_SIZE",
+ parameters = Map(
+ "maxRowSize" -> expectedMaxRowSize.toString,
+ "actualRowSize" -> "1200"))
+ }
+
+ test("SPARK-59824: pickle UDF row-size guard uses SQLConf settings") {
+ val configuredMaxRowSize = 1000L
+ val maxRowHeapFraction = configuredMaxRowSize.toDouble /
Runtime.getRuntime.maxMemory()
+ val expectedMaxRowSize =
+ (Runtime.getRuntime.maxMemory() * maxRowHeapFraction).toLong
+ withSQLConf(
+ SQLConf.PYTHON_UDF_ROW_SIZE_GUARD_ENABLED.key -> "true",
+ SQLConf.PYTHON_UDF_ROW_SIZE_GUARD_MAX_ROW_HEAP_FRACTION.key ->
+ maxRowHeapFraction.toString) {
+ val taskAttr = AttributeReference("s", StringType)()
+ val pythonUdf = PythonUDF(
+ "dummy", new DummyUDF, StringType, Seq(taskAttr),
PythonEvalType.SQL_BATCHED_UDF,
+ udfDeterministic = true)
+ class TestBatchEvalPythonExec extends BatchEvalPythonExec(
+ Seq(pythonUdf),
+ Seq(AttributeReference("result", StringType)()),
+ LocalTableScanExec(Seq(taskAttr), Nil, None)) {
+ def testEvaluatorFactory: BatchEvalPythonEvaluatorFactory =
+ evaluatorFactory.asInstanceOf[BatchEvalPythonEvaluatorFactory]
+ }
+ val plan = new TestBatchEvalPythonExec
+ val oversizedRow = InternalRow(
+ UTF8String.fromString("x" * (expectedMaxRowSize.toInt + 1)))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ plan.testEvaluatorFactory.getInputIterator(
+ Iterator(oversizedRow), StructType.fromDDL("s STRING")).next()
+ },
+ condition = "UDF_LIMITS.ROW_SIZE",
+ parameters = Map(
+ "maxRowSize" -> expectedMaxRowSize.toString,
+ "actualRowSize" -> (expectedMaxRowSize + 1).toString))
+ }
+ }
+
+ test("SPARK-59824: pickle UDF row-size guard disabled no-op and strict
boundary") {
+ val taskAttr = AttributeReference("s", StringType)()
+ val fraction = 1000.toDouble / Runtime.getRuntime.maxMemory()
+ val threshold = (Runtime.getRuntime.maxMemory() * fraction).toLong
+ def checkerFor(
+ enabled: Boolean,
+ maxRowHeapFraction: Double = fraction): Option[InternalRow => Unit] =
+ (new BatchEvalPythonEvaluatorFactory(
+ childOutput = Seq(taskAttr),
+ udfs = Seq.empty,
+ output = Seq.empty,
+ batchSize = 1,
+ maxBytesPerBatch = -1L,
+ pythonMetrics = Map.empty,
+ jobArtifactUUID = None,
+ sessionUUID = None,
+ binaryAsBytes = false,
+ rowSizeGuardEnabled = enabled,
+ rowSizeGuardMaxRowHeapFraction = maxRowHeapFraction) {
+ def testCheck: Option[InternalRow => Unit] =
projectedInputCheck(Seq(StringType))
+ }).testCheck
+ def rowOf(numBytes: Long): InternalRow =
+ InternalRow(UTF8String.fromString("x" * numBytes.toInt))
+ assert(checkerFor(enabled = false).isEmpty)
+
+ val check = checkerFor(enabled = true).get
+ check(rowOf(threshold))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ check(rowOf(threshold + 1))
+ },
+ condition = "UDF_LIMITS.ROW_SIZE",
+ parameters = Map(
+ "maxRowSize" -> threshold.toString,
+ "actualRowSize" -> (threshold + 1).toString))
+
+ // A positive fraction can truncate to a zero-byte limit, which must still
be enforced.
+ val subByteFraction = 0.5 / Runtime.getRuntime.maxMemory()
+ assert((Runtime.getRuntime.maxMemory() * subByteFraction).toLong === 0L)
+ val zeroThresholdCheck = checkerFor(
+ enabled = true, maxRowHeapFraction = subByteFraction).get
+ zeroThresholdCheck(rowOf(0))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ zeroThresholdCheck(rowOf(1))
+ },
+ condition = "UDF_LIMITS.ROW_SIZE",
+ parameters = Map("maxRowSize" -> "0", "actualRowSize" -> "1"))
+ }
+
test("SPARK-57593: ByteBoundedAsArrayIterator oversized-batch and
estimated-bytes metrics") {
def elems(sizes: Long*): Iterator[(Any, Long)] =
sizes.iterator.map(s => (0.asInstanceOf[Any], s))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
