This is an automated email from the ASF dual-hosted git repository.
peter-toth 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 f94586d021ee [SPARK-58578][SQL] Mark ApplyFunctionExpression stateful
to prevent reusedRow data races under concurrent evaluation
f94586d021ee is described below
commit f94586d021ee4e7eff30a85483133ba63fdc74f0
Author: xuanyulu <[email protected]>
AuthorDate: Sun Aug 9 11:46:13 2026 +0200
[SPARK-58578][SQL] Mark ApplyFunctionExpression stateful to prevent
reusedRow data races under concurrent evaluation
### What changes were proposed in this pull request?
Mark `ApplyFunctionExpression` as stateful and add a regression test
verifying
that it produces a fresh expression instance.
### Why are the changes needed?
Sibling QueryExecutions derived from the same base DataFrame can share the
same expression instance. When they execute concurrently, the mutable
reusedRow can therefore be overwritten by another evaluator.
### Does this PR introduce *any* user-facing change?
Yes. It prevents incorrect results during concurrent evaluation of
`ApplyFunctionExpression`. There is no public API change.
### How was this patch tested?
Added `ApplyFunctionExpressionSuite` and ran:
build/sbt 'catalyst/testOnly *ApplyFunctionExpressionSuite'
build/sbt 'catalyst/testOnly *DataSourceV2FunctionSuite'
dev/lint-scala
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #57789 from MrHappyEnding/SPARK-58578.
Authored-by: xuanyulu <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit daba9c6894c2c47a8088c2839938e538db8676ec)
Signed-off-by: Peter Toth <[email protected]>
---
.../expressions/ApplyFunctionExpression.scala | 1 +
.../expressions/ApplyFunctionExpressionSuite.scala | 87 ++++++++++++++++++++++
2 files changed, 88 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpression.scala
index 2cee8303dc57..78f0f8a7af61 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpression.scala
@@ -35,6 +35,7 @@ case class ApplyFunctionExpression(
override lazy val deterministic: Boolean = function.isDeterministic &&
children.forall(_.deterministic)
override def foldable: Boolean = deterministic && children.forall(_.foldable)
+ override def stateful: Boolean = true
private lazy val reusedRow = new
SpecificInternalRow(function.inputTypes().toImmutableArraySeq)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpressionSuite.scala
new file mode 100644
index 000000000000..9300f107a6aa
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ApplyFunctionExpressionSuite.scala
@@ -0,0 +1,87 @@
+/*
+ * 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.catalyst.expressions
+
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.functions.ScalarFunction
+import org.apache.spark.sql.types.{DataType, IntegerType}
+import org.apache.spark.util.ThreadUtils
+
+class ApplyFunctionExpressionSuite extends SparkFunSuite {
+
+ private val intIdentity = new ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "int_identity"
+ override def produceResult(input: InternalRow): Int = input.getInt(0)
+ }
+
+ test("SPARK-58578: ApplyFunctionExpression is stateful and produces fresh
copies") {
+ val expr = ApplyFunctionExpression(
+ intIdentity, Seq(BoundReference(0, IntegerType, nullable = false)))
+ assert(expr.stateful, "ApplyFunctionExpression.stateful should be true")
+ val copy = expr.freshCopyIfContainsStatefulExpression()
+ assert(copy ne expr,
+ "freshCopyIfContainsStatefulExpression should return a new instance " +
+ "for ApplyFunctionExpression")
+ assert(copy.eval(InternalRow(7)) === 7)
+ }
+
+ test("SPARK-58578: fresh copies do not share the reused input row") {
+ val firstEvaluationStarted = new CountDownLatch(1)
+ val secondEvaluationStarted = new CountDownLatch(1)
+ val blockingIdentity = new ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "blocking_identity"
+ override def produceResult(input: InternalRow): Int = {
+ if (input.getInt(0) == 1) {
+ firstEvaluationStarted.countDown()
+ assert(secondEvaluationStarted.await(10, TimeUnit.SECONDS))
+ } else {
+ secondEvaluationStarted.countDown()
+ }
+ input.getInt(0)
+ }
+ }
+
+ val expr = ApplyFunctionExpression(
+ blockingIdentity, Seq(BoundReference(0, IntegerType, nullable = false)))
+ val firstEvaluator = expr.freshCopyIfContainsStatefulExpression()
+ val secondEvaluator = expr.freshCopyIfContainsStatefulExpression()
+
+ val executor = ThreadUtils.newDaemonFixedThreadPool(2,
"apply-function-expression-test")
+ val executionContext = ExecutionContext.fromExecutorService(executor)
+ try {
+ val firstResult =
Future(firstEvaluator.eval(InternalRow(1)))(executionContext)
+ assert(firstEvaluationStarted.await(10, TimeUnit.SECONDS))
+ val secondResult =
Future(secondEvaluator.eval(InternalRow(2)))(executionContext)
+
+ assert(ThreadUtils.awaitResult(firstResult, 10.seconds) === 1)
+ assert(ThreadUtils.awaitResult(secondResult, 10.seconds) === 2)
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
