[SPARK-29419] Seq.toDS / spark.createDataset(Seq) is not thread-safe

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 2.0.2, 2.1.3, 2.2.3, 2.3.4, 2.4.0, 3.0.0
• Fix Version/s: 2.4.6, 3.0.0
• Component/s: SQL
• Labels: correctness
• Target Version/s: 3.0.0
• Assignee: Josh Rosen
• Reporter: Josh Rosen
• Created: 09/Oct/19 20:22
• Updated: 12/Dec/22 18:10
• Resolved: 02/Mar/20 01:20

Description

The Seq.toDS and spark.createDataset(Seq) code is not thread-safe: if the caller-supplied Encoder is used in multiple threads then createDataset's usage of the encoder may lead to incorrect answers because the Encoder's internal mutable state will be updated by from multiple threads.

Here is an example demonstrating the problem:

import org.apache.spark.sql._

val enc = implicitly[Encoder[(Int, Int)]]

val datasets = (1 to 100).par.map { _ =>
  val pairs = (1 to 100).map(x => (x, x))
  spark.createDataset(pairs)(enc)
}

datasets.reduce(_ union _).collect().foreach {
  pair => require(pair._1 == pair._2, s"Pair elements are mismatched: $pair")
}

Due to the thread-safety issue, the above example results in the creation of corrupted records where different input records' fields are co-mingled.

This bug is similar to SPARK-22355, a related problem in Dataset.collect() (fixed in Spark 2.2.1+).

Fortunately, this has a simple one-line fix (copy the encoder); I'll submit a patch for this shortly.

Issue Links

• is related to: SPARK-28456 Add a public API `Encoder.makeCopy` to allow creating Encoder without touching Scala reflections
• relates to: SPARK-22355 Dataset.collect is not threadsafe
• links to: GitHub Pull Request #26076