This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new f25c849381f9 [SPARK-57371][CONNECT] Fix data race in ArrowDeserializers
f25c849381f9 is described below
commit f25c849381f99aefad4fbdc510cd12933f8d3909
Author: Haiyang Sun <[email protected]>
AuthorDate: Thu Jun 18 15:59:58 2026 -0400
[SPARK-57371][CONNECT] Fix data race in ArrowDeserializers
### What changes were proposed in this pull request?
Serialize the Scala runtime-reflection calls in `ArrowDeserializers` so
they are safe under concurrent Arrow deserialization.
While building deserializers, `ArrowDeserializers` uses runtime reflection
in two places:
- resolving a Scala collection's **companion object**
(`mirror.classSymbol(cls).companion.asModule`) to construct the collection's
builder;
- resolving a Scala **`Enumeration`'s module instance**
(`mirror.classSymbol(parent).module.asModule`) to deserialize enum values by
name.
Both now go through a `synchronized` seam on the `ArrowDeserializers`
object monitor, with a consistent two-method shape:
- `resolveCompanion(tag)` -> `resolveCompanionFromMirror(mirror, cls)`
*(synchronized)*
- `resolveEnum(parent)` -> `resolveEnumFromMirror(mirror, parent)`
*(synchronized)*
The thin outer methods read `currentMirror`; the inner `*FromMirror`
methods take the mirror as a parameter and hold the lock around the reflection.
Taking the mirror as a parameter lets the regression test drive these exact
synchronized methods against a deliberately cold mirror, where the race
reproduces. Both methods lock the same monitor, so the two reflection paths
also serialize against each other.
### Why are the changes needed?
Scala runtime reflection is not thread-safe (scala/bug#6240). When several
Arrow result batches are deserialized concurrently against a still-cold
reflection symbol table, `classSymbol(...).companion` / `.module` can observe
the symbol as `NoSymbol`, so `.asModule` fails with:
```
scala.ScalaReflectionException: <none> is not a module
```
The window is normally narrow -- a mirror warms up after the first
resolution of each symbol -- but it is reachable in practice when multiple
deserializers resolve collection/enum types at the same time. Serializing the
reflection through one monitor closes it.
### Does this PR introduce _any_ user-facing change?
No. This is an internal robustness fix; the result of a successful
deserialization is unchanged.
### How was this patch tested?
New `ArrowDeserializersConcurrencySuite` reproduces the race with high
probability: each repetition deterministically re-opens the race window by
building a runtime mirror over a fresh, cold `URLClassLoader` (parented at the
platform loader so `scala.*` is reloaded cold), then drives the real
synchronized reflection from 16 threads x 50 repetitions, covering both
`resolveCompanionFromMirror` (~40 collection companions) and
`resolveEnumFromMirror` (`Enumeration` fixtures). Both tests pass.
Validated end-to-end by toggling the fix:
- **With the fix** -- the concurrency suite is green.
- **Without the fix** (`synchronized` removed) -- the suite goes red with
`scala.ScalaReflectionException: <none> is not a module`, the production
symptom.
- **No functional regression** -- `ArrowEncoderSuite` stays green.
### Was this patch authored or co-authored using generative AI tooling?
YES
Closes #56435 from haiyangsun-db/SPARK-57371.
Authored-by: Haiyang Sun <[email protected]>
Signed-off-by: Herman van Hövell <[email protected]>
(cherry picked from commit fff25e5c43fe65afc4a3f462bc410a571e497edf)
Signed-off-by: Herman van Hövell <[email protected]>
---
.../connect/client/arrow/ArrowEncoderSuite.scala | 75 ++++++++++++++++++++++
.../connect/client/arrow/ArrowDeserializer.scala | 52 +++++++++++++--
2 files changed, 120 insertions(+), 7 deletions(-)
diff --git
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
index 75816a835aaa..6339659a6b0b 100644
---
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
+++
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
@@ -16,15 +16,20 @@
*/
package org.apache.spark.sql.connect.client.arrow
+import java.io.File
import java.math.BigInteger
+import java.net.URLClassLoader
import java.time.{Duration, Period, ZoneOffset}
import java.time.temporal.ChronoUnit
import java.util
import java.util.{Collections, Objects}
+import java.util.concurrent.{ConcurrentLinkedQueue, CyclicBarrier}
import scala.beans.BeanProperty
import scala.collection.mutable
+import scala.jdk.CollectionConverters._
import scala.reflect.classTag
+import scala.reflect.runtime.{universe => ru}
import org.apache.arrow.memory.{BufferAllocator, RootAllocator}
import org.apache.arrow.vector.VarBinaryVector
@@ -1023,6 +1028,76 @@ class ArrowEncoderSuite extends ConnectFunSuite with
BeforeAndAfterAll {
}
}
}
+
+ // SPARK-57371: ArrowDeserializers resolves Scala collection companions and
Enumeration modules
+ // via runtime reflection, which is not thread-safe (scala/bug#6240): a
concurrent
+ // `mirror.classSymbol(cls).companion/.module.asModule` can observe the
symbol as `NoSymbol` and
+ // throw `ScalaReflectionException: <none> is not a module`.
ArrowDeserializers serializes the
+ // reflection through a single monitor. The race only manifests while a
mirror's symbol table is
+ // cold, so each repetition below builds a fresh mirror over a classloader
parented at the
+ // platform loader (so `scala.*` is reloaded cold) and drives the real
synchronized method from
+ // several threads released at once; without the lock it races red.
+
+ private val collectionCompanionClassNames = Seq(
+ "scala.collection.immutable.List",
+ "scala.collection.immutable.Vector",
+ "scala.collection.immutable.Set",
+ "scala.collection.immutable.Map",
+ "scala.collection.mutable.ArrayBuffer",
+ "scala.collection.mutable.HashMap")
+
+ /** A fresh classloader parented at the platform loader, so `scala.*` is
reloaded cold. */
+ private def newColdLoader(): URLClassLoader = {
+ val urls = System
+ .getProperty("java.class.path")
+ .split(File.pathSeparator)
+ .filter(_.nonEmpty)
+ .map(p => new File(p).toURI.toURL)
+ new URLClassLoader(urls, ClassLoader.getPlatformClassLoader)
+ }
+
+ // Drive `resolve` against a fresh cold mirror from 8 threads, 50 times;
fail on any error/hang.
+ private def hammerReflection(
+ names: Seq[String],
+ resolve: (ru.Mirror, Class[_]) => Any): Unit = {
+ val errors = new ConcurrentLinkedQueue[Throwable]()
+ for (_ <- 0 until 50) {
+ val loader = newColdLoader()
+ val mirror = ru.runtimeMirror(loader)
+ val classes = names.map(loader.loadClass)
+ val barrier = new CyclicBarrier(8)
+ val threads = (0 until 8).map { _ =>
+ new Thread(() => {
+ barrier.await() // release all threads simultaneously onto the cold
mirror
+ classes.foreach { cls =>
+ try resolve(mirror, cls)
+ catch { case e: Throwable => errors.add(e) }
+ }
+ })
+ }
+ threads.foreach(_.start())
+ threads.foreach { t =>
+ t.join(60000)
+ assert(!t.isAlive, "thread did not finish within 60s (possible
deadlock)")
+ }
+ }
+ assert(
+ errors.isEmpty,
+ s"reflection raced under concurrent access (${errors.size} error(s)): " +
+ errors.asScala.map(e => s"${e.getClass.getName}:
${e.getMessage}").toSet.mkString("; "))
+ }
+
+ test("SPARK-57371: resolveCompanion is thread-safe under concurrent
cold-mirror access") {
+ hammerReflection(
+ collectionCompanionClassNames,
+ (m, c) => ArrowDeserializers.resolveCompanionFromMirror(m, c))
+ }
+
+ test("SPARK-57371: resolveEnum is thread-safe under concurrent cold-mirror
access") {
+ hammerReflection(
+ Seq(FooEnum.getClass.getName),
+ (m, c) => ArrowDeserializers.resolveEnumFromMirror(m, c))
+ }
}
// TODO fix actual Null fields, e.g.: nullable: Null
diff --git
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
index ceeece073da6..7169fa249b2c 100644
---
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
+++
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
@@ -140,9 +140,9 @@ object ArrowDeserializers {
}
}
case (ScalaEnumEncoder(parent, _), v: FieldVector) =>
- val mirror = scala.reflect.runtime.currentMirror
- val module = mirror.classSymbol(parent).module.asModule
- val enumeration =
mirror.reflectModule(module).instance.asInstanceOf[Enumeration]
+ // Scala runtime reflection is not thread-safe (scala/bug#6240).
Synchronize to
+ // prevent races that surface as "... is not a module" under
concurrent access.
+ val enumeration = resolveEnum(parent)
new LeafFieldDeserializer[Enumeration#Value](encoder, v, timeZoneId) {
override def value(i: Int): Enumeration#Value = {
enumeration.withName(reader.getString(i))
@@ -409,11 +409,49 @@ object ArrowDeserializers {
/**
* Resolve the companion object for a scala class. In our particular case
the class we pass in
* is a Scala collection. We use the companion to create a builder for that
collection.
+ *
+ * Scala runtime reflection is not thread-safe (scala/bug#6240): concurrent
calls to
+ * `classSymbol(...).companion` can race, leaving the companion as
`NoSymbol` so that
+ * `.asModule` throws `ScalaReflectionException: <none> is not a module`. We
serialize the
+ * reflection through this object's monitor (see
[[resolveCompanionFromMirror]]) to prevent it.
+ */
+ private[arrow] def resolveCompanion[T](tag: ClassTag[_]): T =
+ resolveCompanionFromMirror(scala.reflect.runtime.currentMirror,
tag.runtimeClass)
+ .asInstanceOf[T]
+
+ /**
+ * Synchronized reflection to resolve a companion object. The mirror is
passed in rather than
+ * read from `currentMirror` so that the concurrency regression test can
drive this exact
+ * (synchronized) method against a deliberately cold mirror, where the race
would otherwise
+ * surface. Production always passes `currentMirror`.
+ */
+ private[arrow] def resolveCompanionFromMirror(
+ mirror: scala.reflect.runtime.universe.Mirror,
+ cls: Class[_]): Any = synchronized {
+ val module = mirror.classSymbol(cls).companion.asModule
+ mirror.reflectModule(module).instance
+ }
+
+ /**
+ * Resolve a Scala Enumeration parent class to its module instance. Reads
`currentMirror` and
+ * delegates to [[resolveEnumFromMirror]], mirroring the
[[resolveCompanion]] /
+ * [[resolveCompanionFromMirror]] split.
+ */
+ private def resolveEnum(parent: Class[_]): Enumeration =
+ resolveEnumFromMirror(scala.reflect.runtime.currentMirror,
parent).asInstanceOf[Enumeration]
+
+ /**
+ * Synchronized reflection to resolve a Scala Enumeration's module instance.
As with
+ * [[resolveCompanionFromMirror]], the mirror is passed in rather than read
from `currentMirror`
+ * so the concurrency regression test can drive this exact (synchronized)
method against a
+ * deliberately cold mirror. Synchronized on the same monitor as
[[resolveCompanionFromMirror]]
+ * for the same thread-safety reasons. Production always passes
`currentMirror`.
*/
- private[arrow] def resolveCompanion[T](tag: ClassTag[_]): T = {
- val mirror = scala.reflect.runtime.currentMirror
- val module = mirror.classSymbol(tag.runtimeClass).companion.asModule
- mirror.reflectModule(module).instance.asInstanceOf[T]
+ private[arrow] def resolveEnumFromMirror(
+ mirror: scala.reflect.runtime.universe.Mirror,
+ parent: Class[_]): Any = synchronized {
+ val module = mirror.classSymbol(parent).module.asModule
+ mirror.reflectModule(module).instance
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
