HyukjinKwon commented on code in PR #59122:
URL: https://github.com/apache/spark/pull/59122#discussion_r4173320621
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExecSuite.scala:
##########
@@ -81,6 +85,108 @@ 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 conditions = 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 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("NO_ERROR")
+ } catch {
+ case e: SparkThrowable => Iterator(e.getCondition)
+ }
+ }.collect().toSeq
+ assert(conditions === Seq("UDF_LIMITS.ROW_SIZE"))
Review Comment:
**Nit (P3):** These tests check only the error condition (`e.getCondition` /
`conditions === Seq("UDF_LIMITS.ROW_SIZE")`). Since `maxRowSize` and
`actualRowSize` are both `Long` passed via a named map, swapping them at the
`pythonUDFRowSizeExceededError` call site (or in its parameter map) would still
raise the same condition and pass here, while the user-facing message reports
the limit and the actual size swapped. Consider asserting the two message
parameters (e.g. via `checkError`) so the parameter contract has a failure
signal. The current mapping is correct.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala:
##########
@@ -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),
Review Comment:
**Nit (P3):** The guard tests pass `rowSizeGuardEnabled` /
`rowSizeGuardMaxRowHeapFraction` directly to `BatchEvalPythonEvaluatorFactory`,
so this `conf.getConf(...)` read-and-forward is never executed by a test. A
direct swap of the two reads is a compile error (Boolean vs Double), but
dropping a forward (reverting to the factory defaults `false`/`-1.0`) or
reading a wrong same-typed key would silently leave the guard disabled or
misconfigured with no failing test. Consider a test that sets the two SQLConf
entries and drives a pickle UDF through `evaluatorFactory`.
See **Shared repair plan 1** in the review body.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
