You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/03/13 13:34:01 UTC
(spark) branch branch-4.0 updated: [SPARK-51443][SS] Fix singleVariantColumn in DSv2 and readStream
This is an automated email from the ASF dual-hosted git repository.
wenchen 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 d8ac4b7de38c [SPARK-51443][SS] Fix singleVariantColumn in DSv2 and readStream
d8ac4b7de38c is described below
commit d8ac4b7de38cf5c44d1b14fb9142f3a1f48ed97e
Author: Chenhao Li <ch...@databricks.com>
AuthorDate: Thu Mar 13 06:33:36 2025 -0700
[SPARK-51443][SS] Fix singleVariantColumn in DSv2 and readStream
### What changes were proposed in this pull request?
The current JSON `singleVariantColumn` mode doesn't work in DSv2 and `spark.readStream`. This PR fixes the two cases:
- DSv1 calls `JsonFileFormat.inferSchema`, which calls `JsonDataSource.inferSchema`; DSv2 calls `JsonDataSource.inferSchema`. The previous `singleVariantColumn` code was in `JsonFileFormat.inferSchema`, and is now moved into `JsonDataSource.inferSchema`, so that both cases can be covered.
- `spark.readStream` requires that there must be a user-specified schema. `singleVariantColumn` plays the same row as a user-specified schema, but the check would fail.
It also includes a small refactor that moves the option name definition `singleVariantColumn` from `JSONOptions` to `DataSourceOptions`. It will be a common option name shared by multiple data sources (e.g., CSV) when we add the implementation in the future.
### Why are the changes needed?
It is a bug fix that improves the usability of variant.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit test. A test previously in `VariantSuite` is moved to `JsonSuite`, so that we can test the read behavior in both `JsonV1Suite` and `JsonV2Suite`. The test is also extended to include `spark.readStream`.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #50217 from chenhao-db/fix_json_singleVariantColumn.
Authored-by: Chenhao Li <ch...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 71638c4f50d8f5753d4b7bcf157c79395a47dd52)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/DataSourceOptions.scala | 8 ++++
.../spark/sql/catalyst/json/JSONOptions.scala | 2 +-
.../apache/spark/sql/classic/DataFrameReader.scala | 3 +-
.../sql/execution/datasources/DataSource.scala | 6 ++-
.../datasources/json/JsonDataSource.scala | 14 ++++--
.../datasources/json/JsonFileFormat.scala | 5 +-
.../scala/org/apache/spark/sql/VariantSuite.scala | 39 ----------------
.../sql/execution/datasources/json/JsonSuite.scala | 54 ++++++++++++++++++++++
8 files changed, 80 insertions(+), 51 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DataSourceOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DataSourceOptions.scala
index 5348d1054d5d..41b062e4a375 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DataSourceOptions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DataSourceOptions.scala
@@ -64,3 +64,11 @@ trait DataSourceOptions {
*/
def getAlternativeOption(name: String): Option[String] = validOptions.get(name).flatten
}
+
+object DataSourceOptions {
+ // The common option name for all data sources that supports single-variant-column parsing mode.
+ // The option should take in a column name and specifies that the entire record should be stored
+ // as a single VARIANT type column in the table with the given column name.
+ // E.g. spark.read.format("<data-source-format>").option("singleVariantColumn", "colName")
+ val SINGLE_VARIANT_COLUMN = "singleVariantColumn"
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
index 0e1dfdf366a8..249c0efc2681 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
@@ -287,7 +287,7 @@ object JSONOptions extends DataSourceOptions {
val COLUMN_NAME_OF_CORRUPTED_RECORD = newOption("columnNameOfCorruptRecord")
val TIME_ZONE = newOption("timeZone")
val WRITE_NON_ASCII_CHARACTER_AS_CODEPOINT = newOption("writeNonAsciiCharacterAsCodePoint")
- val SINGLE_VARIANT_COLUMN = newOption("singleVariantColumn")
+ val SINGLE_VARIANT_COLUMN = newOption(DataSourceOptions.SINGLE_VARIANT_COLUMN)
val USE_UNSAFE_ROW = newOption("useUnsafeRow")
// Options with alternative
val ENCODING = "encoding"
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
index 489ac31e5291..08cc82375226 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
@@ -27,6 +27,7 @@ import org.apache.spark.api.java.JavaRDD
import org.apache.spark.rdd.RDD
import org.apache.spark.sql
import org.apache.spark.sql.Encoders
+import org.apache.spark.sql.catalyst.DataSourceOptions
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.csv.{CSVHeaderChecker, CSVOptions, UnivocityParser}
import org.apache.spark.sql.catalyst.expressions.ExprUtils
@@ -336,7 +337,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession)
/** @inheritdoc */
override protected def validateSingleVariantColumn(): Unit = {
- if (extraOptions.get(JSONOptions.SINGLE_VARIANT_COLUMN).isDefined &&
+ if (extraOptions.get(DataSourceOptions.SINGLE_VARIANT_COLUMN).isDefined &&
userSpecifiedSchema.isDefined) {
throw QueryCompilationErrors.invalidSingleVariantColumn()
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 97c88d660b00..0b8346920baa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -30,6 +30,7 @@ import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys.{CLASS_NAME, DATA_SOURCE, DATA_SOURCES, PATHS}
import org.apache.spark.sql.{AnalysisException, SaveMode, SparkSession}
+import org.apache.spark.sql.catalyst.DataSourceOptions
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogTable, CatalogUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -261,8 +262,11 @@ case class DataSource(
val isSchemaInferenceEnabled = sparkSession.sessionState.conf.streamingSchemaInference
val isTextSource = providingClass == classOf[text.TextFileFormat]
+ val isSingleVariantColumn = providingClass == classOf[json.JsonFileFormat] &&
+ caseInsensitiveOptions.contains(DataSourceOptions.SINGLE_VARIANT_COLUMN)
// If the schema inference is disabled, only text sources require schema to be specified
- if (!isSchemaInferenceEnabled && !isTextSource && userSpecifiedSchema.isEmpty) {
+ if (!isSchemaInferenceEnabled && !isTextSource && !isSingleVariantColumn &&
+ userSpecifiedSchema.isEmpty) {
throw QueryExecutionErrors.createStreamingSourceNotSpecifySchemaError()
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
index 3a4ca99fc95a..bedf5ec62e4e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.classic.ClassicConversions.castToImpl
import org.apache.spark.sql.execution.SQLExecution
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.text.TextFileFormat
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.Utils
@@ -61,10 +61,14 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): Option[StructType] = {
- if (inputPaths.nonEmpty) {
- Some(infer(sparkSession, inputPaths, parsedOptions))
- } else {
- None
+ parsedOptions.singleVariantColumn match {
+ case Some(columnName) => Some(StructType(Array(StructField(columnName, VariantType))))
+ case None =>
+ if (inputPaths.nonEmpty) {
+ Some(infer(sparkSession, inputPaths, parsedOptions))
+ } else {
+ None
+ }
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
index 6174c017f604..571bb13a0983 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
@@ -55,10 +55,7 @@ class JsonFileFormat extends TextBasedFileFormat with DataSourceRegister {
options,
sparkSession.sessionState.conf.sessionLocalTimeZone,
sparkSession.sessionState.conf.columnNameOfCorruptRecord)
- parsedOptions.singleVariantColumn match {
- case Some(columnName) => Some(StructType(Array(StructField(columnName, VariantType))))
- case None => JsonDataSource(parsedOptions).inferSchema(sparkSession, files, parsedOptions)
- }
+ JsonDataSource(parsedOptions).inferSchema(sparkSession, files, parsedOptions)
}
override def prepareWrite(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
index b6fe4af28ab0..e57ff2f3d236 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
@@ -453,45 +453,6 @@ class VariantSuite extends QueryTest with SharedSparkSession with ExpressionEval
}
}
- test("json scan") {
- val content = Seq(
- "true",
- """{"a": [], "b": null}""",
- """{"a": 1}""",
- "[1, 2, 3]"
- ).mkString("\n").getBytes(StandardCharsets.UTF_8)
-
- withTempDir { dir =>
- val file = new File(dir, "file.json")
- Files.write(file.toPath, content)
-
- checkAnswer(
- spark.read.format("json").option("singleVariantColumn", "var")
- .load(file.getAbsolutePath)
- .selectExpr("to_json(var)"),
- Seq(Row("true"), Row("""{"a":[],"b":null}"""), Row("""{"a":1}"""), Row("[1,2,3]"))
- )
-
- checkAnswer(
- spark.read.format("json").schema("a variant, b variant")
- .load(file.getAbsolutePath).selectExpr("to_json(a)", "to_json(b)"),
- Seq(Row(null, null), Row("[]", "null"), Row("1", null), Row(null, null))
- )
- }
-
- // Test scan with partitions.
- withTempDir { dir =>
- new File(dir, "a=1/b=2/").mkdirs()
- Files.write(new File(dir, "a=1/b=2/file.json").toPath, content)
- checkAnswer(
- spark.read.format("json").option("singleVariantColumn", "var")
- .load(dir.getAbsolutePath).selectExpr("a", "b", "to_json(var)"),
- Seq(Row(1, 2, "true"), Row(1, 2, """{"a":[],"b":null}"""), Row(1, 2, """{"a":1}"""),
- Row(1, 2, "[1,2,3]"))
- )
- }
- }
-
test("json scan with map schema") {
withTempDir { dir =>
val file = new File(dir, "file.json")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index d79709317c3d..07fb2dedaa0a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -4001,6 +4001,60 @@ abstract class JsonSuite
"{\"key\":90000001,\"value\":\"Hey there\"}"
)
}
+
+ test("json with variant") {
+ val content = Seq(
+ "true",
+ """{"a": [], "b": null}""",
+ """{"a": 1}""",
+ "[1, 2, 3]"
+ ).mkString("\n").getBytes(StandardCharsets.UTF_8)
+
+ withTempDir { dir =>
+ val file = new File(dir, "file.json")
+ Files.write(file.toPath, content)
+
+ checkAnswer(
+ spark.read.format("json").option("singleVariantColumn", "var")
+ .load(file.getAbsolutePath)
+ .selectExpr("to_json(var)"),
+ Seq(Row("true"), Row("""{"a":[],"b":null}"""), Row("""{"a":1}"""), Row("[1,2,3]"))
+ )
+
+ checkAnswer(
+ spark.read.format("json").schema("a variant, b variant")
+ .load(file.getAbsolutePath).selectExpr("to_json(a)", "to_json(b)"),
+ Seq(Row(null, null), Row("[]", "null"), Row("1", null), Row(null, null))
+ )
+
+ withTempDir { streamDir =>
+ val stream = spark.readStream.format("json").option("singleVariantColumn", "var")
+ .load(dir.getCanonicalPath)
+ .selectExpr("to_json(var)")
+ .writeStream
+ .option("checkpointLocation", streamDir.getCanonicalPath + "/checkpoint")
+ .format("parquet")
+ .start(streamDir.getCanonicalPath + "/output")
+ stream.processAllAvailable()
+ checkAnswer(
+ spark.read.format("parquet").load(streamDir.getCanonicalPath + "/output"),
+ Seq(Row("true"), Row("""{"a":[],"b":null}"""), Row("""{"a":1}"""), Row("[1,2,3]"))
+ )
+ }
+ }
+
+ // Test scan with partitions.
+ withTempDir { dir =>
+ new File(dir, "a=1/b=2/").mkdirs()
+ Files.write(new File(dir, "a=1/b=2/file.json").toPath, content)
+ checkAnswer(
+ spark.read.format("json").option("singleVariantColumn", "var")
+ .load(dir.getAbsolutePath).selectExpr("a", "b", "to_json(var)"),
+ Seq(Row(1, 2, "true"), Row(1, 2, """{"a":[],"b":null}"""), Row(1, 2, """{"a":1}"""),
+ Row(1, 2, "[1,2,3]"))
+ )
+ }
+ }
}
class JsonV1Suite extends JsonSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org