This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 89d5f1898b32 [SPARK-57678][SQL] Add explodeEmbeddedArray JSON option
for streaming scans of embedded arrays.
89d5f1898b32 is described below
commit 89d5f1898b325d4d719a075b8f6140fb4c08c639
Author: chenhao-db <[email protected]>
AuthorDate: Wed Jul 1 09:28:30 2026 +0800
[SPARK-57678][SQL] Add explodeEmbeddedArray JSON option for streaming scans
of embedded arrays.
### What changes were proposed in this pull request?
Adds a new JSON reader option `explodeEmbeddedArray` for reading JSON files
that are object documents with a top-level array-valued field, e.g. `{...,
"Records": [record1, record2, ...], ...}`. The option value names the array
field to explode; each element of that array becomes one row, stored as a
single VARIANT column (like `singleVariantColumn`, with which it is mutually
exclusive). The inferred schema names the variant column after the array field,
and a user-specified schema may [...]
The records are split out of the array by a new streaming
`EmbeddedArraySplitter`, and each record is then parsed by the unchanged JSON
parsing logic, so neither the whole array text nor the result rows are ever
buffered in memory. The scan is whole-file: a new `EmbeddedArrayJsonDataSource`
reports `isSplitable = false`, which both the V1 `JsonFileFormat` and the V2
`JsonScan` already consult, so the option works in both v1 and v2 scans
(including compressed files, partitions, and str [...]
It is an intentional design decision that anything outside the array is
ignored, because:
1. Users only care about the records array in common use cases (e.g. AWS
CloudTrail).
2. It is difficult to implement if we want to achive both: avoid buffering
the whole array, scanning the file only once.
### Why are the changes needed?
Some common JSON formats (e.g. AWS CloudTrail) wrap their records in a
single top-level array inside one large object document. Today, ingesting such
files requires materializing the whole array, which requires all elements to
stay in memory at the same time and can easily cause OOM. This option makes
that ingestion efficient by streaming the records out one at a time as if they
were independent input records, never buffering the array.
### Does this PR introduce _any_ user-facing change?
Yes — a new opt-in JSON reader option `explodeEmbeddedArray`. There is no
behavior change unless the option is set. Example:
```scala
// file.json: {"Records": [{"a": 1}, {"b": 2}]}
spark.read.format("json").option("explodeEmbeddedArray",
"Records").load(path)
// => 2 rows, one VARIANT column named "Records": {"a":1}, {"b":2}
```
It also adds three new error conditions:
`EXPLODE_EMBEDDED_ARRAY_CONFLICTING_OPTION`,
`EXPLODE_EMBEDDED_ARRAY_UNSUPPORTED_USAGE`, and
`INVALID_EXPLODE_EMBEDDED_ARRAY_SCHEMA`.
### How was this patch tested?
New `EmbeddedArraySplitterSuite` (splitter unit tests, including custom
array field names, nesting/escapes, scalars, BOM, truncated input, and inputs
larger than the internal buffer) and
`ExplodeEmbeddedArrayJsonV1Suite`/`ExplodeEmbeddedArrayJsonV2Suite`
(end-to-end: schema inference/validation, array field names other than
`Records`, user schemas that rename the variant column, malformed records in
all parse modes, truncated/compressed/partitioned/streaming inputs, option
conflicts, [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56756 from chenhao-db/explodeEmbeddedArray.
Authored-by: chenhao-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 19 +
docs/sql-data-sources-json.md | 6 +
.../spark/sql/catalyst/DataSourceOptions.scala | 33 ++
.../sql/catalyst/expressions/jsonExpressions.scala | 25 +-
.../spark/sql/catalyst/json/JSONOptions.scala | 23 ++
.../spark/sql/catalyst/json/JacksonParser.scala | 5 +
.../spark/sql/errors/QueryCompilationErrors.scala | 18 +
.../apache/spark/sql/classic/DataFrameReader.scala | 11 +-
.../sql/execution/datasources/DataSource.scala | 6 +-
.../json/EmbeddedArrayJsonDataSource.scala | 319 +++++++++++++++++
.../datasources/json/JsonDataSource.scala | 6 +-
.../json/ExplodeEmbeddedArrayJsonSuite.scala | 396 +++++++++++++++++++++
.../sql/execution/datasources/json/JsonSuite.scala | 3 +-
13 files changed, 857 insertions(+), 13 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index a4450e5317b6..3818b2d4b53a 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -2457,6 +2457,19 @@
},
"sqlState" : "42809"
},
+ "EXPLODE_EMBEDDED_ARRAY_CONFLICTING_OPTION" : {
+ "message" : [
+ "The 'explodeEmbeddedArray' DataFrame API reader option is mutually
exclusive with the '<option>' DataFrame API option.",
+ "Please remove one of them and then retry the DataFrame operation again."
+ ],
+ "sqlState" : "4274K"
+ },
+ "EXPLODE_EMBEDDED_ARRAY_UNSUPPORTED_USAGE" : {
+ "message" : [
+ "The 'explodeEmbeddedArray' option can only be used when reading JSON
files. It is not supported in <usage>."
+ ],
+ "sqlState" : "0A000"
+ },
"EXPRESSION_DECODING_FAILED" : {
"message" : [
"Failed to decode a row to a value of the expressions: <expressions>."
@@ -3823,6 +3836,12 @@
],
"sqlState" : "F0000"
},
+ "INVALID_EXPLODE_EMBEDDED_ARRAY_SCHEMA" : {
+ "message" : [
+ "User specified schema <schema> is invalid when the
`explodeEmbeddedArray` option is enabled. The schema must either be a variant
field, or a variant field plus a corrupt column field."
+ ],
+ "sqlState" : "42613"
+ },
"INVALID_EXPRESSION_ENCODER" : {
"message" : [
"Found an invalid expression encoder. Expects an instance of
ExpressionEncoder but got <encoderType>. For more information consult
'<docroot>/api/java/index.html?org/apache/spark/sql/Encoder.html'."
diff --git a/docs/sql-data-sources-json.md b/docs/sql-data-sources-json.md
index 19e724deb0bf..ba8691320f90 100644
--- a/docs/sql-data-sources-json.md
+++ b/docs/sql-data-sources-json.md
@@ -219,6 +219,12 @@ Data source options of JSON can be set via:
<td>If specified, the entire JSON record is parsed and stored as a single
column of <code>VariantType</code> with the given column name, instead of being
split into individual fields.</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>explodeEmbeddedArray</code></td>
+ <td>(none)</td>
+ <td>If specified, each input file is treated as a single JSON object
document with a top-level array-valued field of the given name (e.g.
<code>{"Records": [record1, record2, ...]}</code>). Each element of that array
becomes one row, stored as a single column of <code>VariantType</code>, and
anything outside the array is ignored. This is useful for ingesting formats
such as AWS CloudTrail. Mutually exclusive with
<code>singleVariantColumn</code>. Because each file is consumed as a wh [...]
+ <td>read</td>
+ </tr>
<tr>
<td><code>enableDateTimeParsingFallback</code></td>
<td>Enabled if the time parser policy has legacy settings or if no custom
date or timestamp pattern was provided.</td>
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 aa5e3c1de13f..0595eef2b8e8 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
@@ -84,6 +84,10 @@ object DataSourceOptions {
// the corrupt record will be dropped in DROPMALFORMED mode.
val COLUMN_NAME_OF_CORRUPT_RECORD = "columnNameOfCorruptRecord"
+ // The JSON data source option for exploding an embedded top-level array
into rows. See
+ // `JSONOptions.explodeEmbeddedArray` for details.
+ val EXPLODE_EMBEDDED_ARRAY = "explodeEmbeddedArray"
+
// When `singleVariantColumn` is enabled and there is a user-specified
schema, the schema must
// either be a variant field, or a variant field plus a corrupt column field.
def validateSingleVariantColumn(
@@ -111,4 +115,33 @@ object DataSourceOptions {
case _ =>
}
}
+
+ // When `explodeEmbeddedArray` is enabled, a user-specified schema has the
same requirement as
+ // `singleVariantColumn`, except that the variant field may have any name
(the option value
+ // names the embedded array field, not the column): either a single variant
field, or that plus
+ // a corrupt record field. Keep the logic in sync with
`validateSingleVariantColumn`.
+ def validateExplodeEmbeddedArray(
+ options: CaseInsensitiveMap[String],
+ userSpecifiedSchema: Option[StructType]): Unit = {
+ (options.get(EXPLODE_EMBEDDED_ARRAY), userSpecifiedSchema) match {
+ case (Some(_), Some(schema)) =>
+ val corruptRecordColumnName = options.getOrElse(
+ COLUMN_NAME_OF_CORRUPT_RECORD, SQLConf.get.columnNameOfCorruptRecord)
+ var valid = schema.fields.count { f =>
+ f.dataType.isInstanceOf[VariantType] && f.name !=
corruptRecordColumnName && f.nullable
+ } == 1
+ schema.length match {
+ case 1 =>
+ case 2 =>
+ valid = valid && schema.fields.exists { f =>
+ f.dataType.isInstanceOf[StringType] && f.name ==
corruptRecordColumnName && f.nullable
+ }
+ case _ => valid = false
+ }
+ if (!valid) {
+ throw
QueryCompilationErrors.invalidExplodeEmbeddedArraySchema(schema)
+ }
+ case _ =>
+ }
+ }
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index b82ed39824f9..471916c68860 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -29,6 +29,7 @@ import
org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.json._
import org.apache.spark.sql.catalyst.trees.TreePattern.{GET_JSON_OBJECT,
JSON_TO_STRUCT,
RUNTIME_REPLACEABLE, TreePattern}
+import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
@@ -384,14 +385,22 @@ case class JsonToStructs(
child = child,
timeZoneId = None)
- override def checkInputDataTypes(): TypeCheckResult = nullableSchema match {
- case _: StructType | _: ArrayType | _: MapType | _: VariantType =>
- val checkResult = ExprUtils.checkJsonSchema(nullableSchema)
- if (checkResult.isFailure) checkResult else super.checkInputDataTypes()
- case _ =>
- DataTypeMismatch(
- errorSubClass = "INVALID_JSON_SCHEMA",
- messageParameters = Map("schema" -> toSQLType(nullableSchema)))
+ override def checkInputDataTypes(): TypeCheckResult = {
+ // `from_json` parses each input string as one JSON document, so the
embedded array
+ // splitting can never apply.
+ if
(CaseInsensitiveMap(options).contains(JSONOptions.EXPLODE_EMBEDDED_ARRAY)) {
+ throw QueryCompilationErrors.explodeEmbeddedArrayUnsupportedUsage(
+ "the from_json function")
+ }
+ nullableSchema match {
+ case _: StructType | _: ArrayType | _: MapType | _: VariantType =>
+ val checkResult = ExprUtils.checkJsonSchema(nullableSchema)
+ if (checkResult.isFailure) checkResult else super.checkInputDataTypes()
+ case _ =>
+ DataTypeMismatch(
+ errorSubClass = "INVALID_JSON_SCHEMA",
+ messageParameters = Map("schema" -> toSQLType(nullableSchema)))
+ }
}
override def dataType: DataType = nullableSchema
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 2fcba9c29263..c7e8a8e13116 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
@@ -27,6 +27,7 @@ import com.fasterxml.jackson.core.json.JsonReadFeature
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
/**
@@ -214,6 +215,27 @@ class JSONOptions(
// E.g. spark.read.format("json").option("singleVariantColumn", "colName")
val singleVariantColumn: Option[String] =
parameters.get(SINGLE_VARIANT_COLUMN)
+ // This option takes in a field name and specifies that the input JSON files
are object
+ // documents with a top-level array-valued field of that name, i.e.
+ // `{..., "<fieldName>": [element1, element2, ...], ...}`. Each element of
the array becomes one
+ // row, stored as a single VARIANT type column. Anything outside the array
is ignored. The
+ // elements are streamed out of the array one at a time, without ever
buffering the whole
+ // document; the option intends to make ingestion of formats like AWS
CloudTrail
+ // (`{"Records": [...]}`) more efficient.
+ // The inferred schema names the variant column after the array field; a
user-specified schema
+ // may give the column any name. The file is always consumed as a whole
document, so its line
+ // layout is irrelevant and line-based options such as `multiLine` and
`lineSep` are ignored.
+ // Unlike the other JSON readers, which let Jackson auto-detect the charset
of the byte stream,
+ // this reader decodes the file to characters itself and so reads it as
UTF-8 unless the
+ // `encoding` option specifies another charset.
+ // Similar to `singleVariantColumn`; the two options are mutually exclusive.
+ val explodeEmbeddedArray: Option[String] =
parameters.get(EXPLODE_EMBEDDED_ARRAY)
+
+ if (explodeEmbeddedArray.isDefined && singleVariantColumn.isDefined) {
+ throw QueryCompilationErrors.explodeEmbeddedArrayConflictingOption(
+ DataSourceOptions.SINGLE_VARIANT_COLUMN)
+ }
+
val useUnsafeRow: Boolean =
parameters.get(USE_UNSAFE_ROW).map(_.toBoolean).getOrElse(
SQLConf.get.getConf(SQLConf.JSON_USE_UNSAFE_ROW))
@@ -314,6 +336,7 @@ object JSONOptions extends DataSourceOptions {
val TIME_ZONE = newOption("timeZone")
val WRITE_NON_ASCII_CHARACTER_AS_CODEPOINT =
newOption("writeNonAsciiCharacterAsCodePoint")
val SINGLE_VARIANT_COLUMN =
newOption(DataSourceOptions.SINGLE_VARIANT_COLUMN)
+ val EXPLODE_EMBEDDED_ARRAY =
newOption(DataSourceOptions.EXPLODE_EMBEDDED_ARRAY)
val USE_UNSAFE_ROW = newOption("useUnsafeRow")
// Options with alternative
val ENCODING = "encoding"
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index f5d90c5a09bf..ea8061b774c4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -115,6 +115,11 @@ class JacksonParser(
case _: StructType if options.singleVariantColumn.isDefined => (parser:
JsonParser) => {
Some(InternalRow(parseVariant(parser)))
}
+ // Each embedded array element is parsed into a single variant column,
like
+ // `singleVariantColumn`. The elements have already been split out by
the data source.
+ case _: StructType if options.explodeEmbeddedArray.isDefined => (parser:
JsonParser) => {
+ Some(InternalRow(parseVariant(parser)))
+ }
case st: StructType => makeStructRootConverter(st)
case mt: MapType => makeMapRootConverter(mt)
case at: ArrayType => makeArrayRootConverter(at)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index b54c21e2a5da..30cbad712d35 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -3790,6 +3790,24 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("schema" -> toSQLType(schema)))
}
+ def invalidExplodeEmbeddedArraySchema(schema: DataType): Throwable = {
+ new AnalysisException(
+ errorClass = "INVALID_EXPLODE_EMBEDDED_ARRAY_SCHEMA",
+ messageParameters = Map("schema" -> toSQLType(schema)))
+ }
+
+ def explodeEmbeddedArrayConflictingOption(option: String): Throwable = {
+ new AnalysisException(
+ errorClass = "EXPLODE_EMBEDDED_ARRAY_CONFLICTING_OPTION",
+ messageParameters = Map("option" -> option))
+ }
+
+ def explodeEmbeddedArrayUnsupportedUsage(usage: String): Throwable = {
+ new AnalysisException(
+ errorClass = "EXPLODE_EMBEDDED_ARRAY_UNSUPPORTED_USAGE",
+ messageParameters = Map("usage" -> usage))
+ }
+
def writeWithSaveModeUnsupportedBySourceError(source: String, createMode:
String): Throwable = {
new AnalysisException(
errorClass = "UNSUPPORTED_DATA_SOURCE_SAVE_MODE",
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 3dbdf0530516..5ba6cb204075 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
@@ -37,6 +37,7 @@ import
org.apache.spark.sql.catalyst.plans.logical.UnresolvedDataSource
import org.apache.spark.sql.catalyst.util.FailureSafeParser
import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, XmlOptions}
import org.apache.spark.sql.classic.ClassicConversions._
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.csv._
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions,
JDBCPartition, JDBCRelation}
import
org.apache.spark.sql.execution.datasources.json.JsonUtils.checkJsonSchema
@@ -166,6 +167,12 @@ class DataFrameReader private[sql](sparkSession:
SparkSession)
sparkSession.sessionState.conf.sessionLocalTimeZone,
sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ // Each element of the input dataset is already one JSON document, so the
embedded array
+ // splitting can never apply.
+ if (parsedOptions.explodeEmbeddedArray.isDefined) {
+ throw QueryCompilationErrors.explodeEmbeddedArrayUnsupportedUsage(
+ "the json(Dataset[String]) API")
+ }
userSpecifiedSchema.foreach(checkJsonSchema)
val schema = userSpecifiedSchema.map {
case s if !SQLConf.get.getConf(
@@ -349,8 +356,10 @@ class DataFrameReader private[sql](sparkSession:
SparkSession)
override def textFile(paths: String*): Dataset[String] =
super.textFile(paths: _*)
/** @inheritdoc */
- override protected def validateSingleVariantColumn(): Unit =
+ override protected def validateSingleVariantColumn(): Unit = {
DataSourceOptions.validateSingleVariantColumn(extraOptions,
userSpecifiedSchema)
+ DataSourceOptions.validateExplodeEmbeddedArray(extraOptions,
userSpecifiedSchema)
+ }
override protected def validateJsonSchema(): Unit =
userSpecifiedSchema.foreach(checkJsonSchema)
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 e249e8cfa4cf..8336e13b59e1 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
@@ -276,9 +276,13 @@ case class DataSource(
val isSingleVariantColumn = (providingClass ==
classOf[json.JsonFileFormat] ||
providingClass == classOf[csv.CSVFileFormat]) &&
caseInsensitiveOptions.contains(DataSourceOptions.SINGLE_VARIANT_COLUMN)
+ // Like `singleVariantColumn`, an embedded-array JSON scan has a fixed
schema and does
+ // not require schema inference.
+ val isExplodeEmbeddedArray = providingClass ==
classOf[json.JsonFileFormat] &&
+
caseInsensitiveOptions.contains(DataSourceOptions.EXPLODE_EMBEDDED_ARRAY)
// If the schema inference is disabled, only text sources require
schema to be specified
if (!isSchemaInferenceEnabled && !isTextSource &&
!isSingleVariantColumn &&
- userSpecifiedSchema.isEmpty) {
+ !isExplodeEmbeddedArray && userSpecifiedSchema.isEmpty) {
throw
QueryExecutionErrors.createStreamingSourceNotSpecifySchemaError()
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/EmbeddedArrayJsonDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/EmbeddedArrayJsonDataSource.scala
new file mode 100644
index 000000000000..f5e96d669d1e
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/EmbeddedArrayJsonDataSource.scala
@@ -0,0 +1,319 @@
+/*
+ * 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.execution.datasources.json
+
+import java.io.{InputStream, InputStreamReader, Reader}
+import java.nio.charset.StandardCharsets
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileStatus
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser,
JSONOptions}
+import org.apache.spark.sql.catalyst.util.FailureSafeParser
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.datasources.{CodecStreams,
PartitionedFile}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * A [[JsonDataSource]] for the `explodeEmbeddedArray` option. The input JSON
files are expected
+ * to be object documents with a top-level array-valued field named by the
option, i.e.
+ * `{..., "<fieldName>": [record1, record2, ...], ...}`. An
[[EmbeddedArraySplitter]] streams the
+ * records out of the array one at a time, and each record is parsed by the
regular JSON parsing
+ * logic as if it were an independent input record, producing one row per
record. Neither the
+ * whole array text nor the result rows are ever buffered in memory.
+ */
+object EmbeddedArrayJsonDataSource extends JsonDataSource {
+ // The embedded array spans the whole file, so the file cannot be split.
+ override val isSplitable: Boolean = false
+
+ override protected def infer(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ // Unreachable: `inferSchema` always returns a single variant column when
+ // `explodeEmbeddedArray` is set.
+ throw SparkException.internalError(
+ "The schema is always a single variant column when
`explodeEmbeddedArray` is set.")
+ }
+
+ override def readFile(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] = {
+ val stream = CodecStreams.createInputStreamWithCloseResource(conf,
file.toPath)
+ parseStream(stream, parser, schema)
+ }
+
+ override protected def readStream(
+ in: InputStream,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] = {
+ parseStream(in, parser, schema)
+ }
+
+ private def parseStream(
+ stream: InputStream,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] = {
+ // The splitter scans characters, so it needs a decoded `Reader`. Unlike
the other readers --
+ // which hand the byte stream to Jackson and let it auto-detect the
charset -- we must pick the
+ // charset up front, so a non-UTF-8 file requires the `encoding` option to
be set explicitly.
+ val encoding =
parser.options.encoding.getOrElse(StandardCharsets.UTF_8.name())
+ val splitter = new EmbeddedArraySplitter(
+ new InputStreamReader(stream, encoding),
parser.options.explodeEmbeddedArray.get)
+ val records = new Iterator[String] {
+ private[this] var nextRecord: Option[String] = None
+
+ override def hasNext: Boolean = {
+ if (nextRecord.isEmpty) {
+ nextRecord = splitter.nextRecord()
+ }
+ nextRecord.isDefined
+ }
+
+ override def next(): String = {
+ if (!hasNext) {
+ throw QueryExecutionErrors.endOfStreamError()
+ }
+ val record = nextRecord.get
+ nextRecord = None
+ record
+ }
+ }
+ val safeParser = new FailureSafeParser[String](
+ input => parser.parse(input, CreateJacksonParser.string,
UTF8String.fromString),
+ parser.options.parseMode,
+ schema,
+ parser.options.columnNameOfCorruptRecord)
+ records.flatMap(safeParser.parse)
+ }
+}
+
+/**
+ * Splits the array value of the top-level field `arrayFieldName` of a JSON
object document
+ * (`{..., "<arrayFieldName>": [record1, record2, ...], ...}`) into individual
record texts. Only
+ * one record is held in memory at a time; the whole array is never
materialized. Buffering the
+ * record text does not change the memory bound: each record is parsed into a
variant value that
+ * must be fully materialized in the result row anyway, so the peak memory
usage is O(record
+ * size) either way.
+ *
+ * The splitter only understands enough JSON structure (strings, escapes,
nesting) to find the
+ * record boundaries. The returned record text is not validated here;
malformed records are
+ * rejected later by the regular JSON parser. Anything outside the array is
ignored, and a
+ * document without a matching top-level array field yields no records. If the
document ends in
+ * the middle of a record, the partial record text is returned and left for
the JSON parser to
+ * reject.
+ */
+class EmbeddedArraySplitter(reader: Reader, arrayFieldName: String) {
+ private[this] val buffer = new Array[Char](EmbeddedArraySplitter.BUFFER_SIZE)
+ private[this] var bufferLen = 0
+ private[this] var pos = 0
+ private[this] var initialized = false
+ // Whether we are positioned inside the embedded array with records
potentially remaining.
+ private[this] var inArray = false
+
+ /** Returns the text of the next record, or None when there are no more
records. */
+ def nextRecord(): Option[String] = {
+ if (!initialized) {
+ initialized = true
+ inArray = findEmbeddedArray()
+ }
+ if (!inArray) {
+ return None
+ }
+ var c = nextNonWhitespace()
+ // Tolerate stray commas between records rather than producing garbage
records.
+ while (c == ',') {
+ c = nextNonWhitespace()
+ }
+ if (c == -1 || c == ']') {
+ inArray = false
+ return None
+ }
+ val out = new StringBuilder
+ consumeValue(c, out)
+ Some(out.toString)
+ }
+
+ /**
+ * Scans the top-level JSON object until the reader is positioned inside the
embedded array,
+ * just after its opening '['. Returns false if there is no top-level array
field named
+ * `arrayFieldName`.
+ */
+ private def findEmbeddedArray(): Boolean = {
+ var c = nextNonWhitespace()
+ if (c == 0xFEFF) { // skip the byte order mark
+ c = nextNonWhitespace()
+ }
+ if (c != '{') {
+ return false
+ }
+ while (true) {
+ c = nextNonWhitespace()
+ while (c == ',') {
+ c = nextNonWhitespace()
+ }
+ if (c != '"') {
+ // '}', EOF, or malformed content: in all cases there is no embedded
array ahead.
+ return false
+ }
+ val key = readKey()
+ c = nextNonWhitespace()
+ if (c != ':') {
+ return false
+ }
+ c = nextNonWhitespace()
+ if (c == -1) {
+ return false
+ }
+ if (key == arrayFieldName && c == '[') {
+ return true
+ }
+ consumeValue(c, out = null)
+ }
+ false // unreachable
+ }
+
+ /**
+ * Reads a JSON object key after its opening quote and returns the raw key
text. Escape
+ * sequences are not processed: the key is matched against `arrayFieldName`
by raw text, so a
+ * key written with escape sequences in the document does not match.
+ */
+ private def readKey(): String = {
+ val out = new StringBuilder
+ consumeString(out)
+ if (out.nonEmpty && out.charAt(out.length - 1) == '"') {
+ out.setLength(out.length - 1)
+ }
+ out.toString
+ }
+
+ /**
+ * Consumes one JSON value whose first character (already consumed) is
`first`, appending the
+ * value text to `out` unless it is null. The trailing delimiter (',', ']',
'}' or whitespace)
+ * is left unconsumed.
+ */
+ private def consumeValue(first: Int, out: StringBuilder): Unit = {
+ if (out != null) {
+ out.append(first.toChar)
+ }
+ if (first == '"') {
+ consumeString(out)
+ } else if (first == '{' || first == '[') {
+ var depth = 1
+ while (depth > 0) {
+ val c = nextChar()
+ if (c == -1) {
+ return
+ }
+ if (out != null) {
+ out.append(c.toChar)
+ }
+ if (c == '"') {
+ consumeString(out)
+ } else if (c == '{' || c == '[') {
+ depth += 1
+ } else if (c == '}' || c == ']') {
+ depth -= 1
+ }
+ }
+ } else {
+ // A scalar (number, true, false, null): consume until a delimiter.
+ var c = peekChar()
+ while (c != -1 && c != ',' && c != '}' && c != ']' && !isWhitespace(c)) {
+ if (out != null) {
+ out.append(c.toChar)
+ }
+ nextChar()
+ c = peekChar()
+ }
+ }
+ }
+
+ /**
+ * Consumes a JSON string after its opening quote, appending the consumed
characters (including
+ * the closing quote) to `out` unless it is null.
+ */
+ private def consumeString(out: StringBuilder): Unit = {
+ while (true) {
+ val c = nextChar()
+ if (c == -1) {
+ return
+ }
+ if (out != null) {
+ out.append(c.toChar)
+ }
+ if (c == '\\') {
+ val escaped = nextChar()
+ if (escaped != -1 && out != null) {
+ out.append(escaped.toChar)
+ }
+ } else if (c == '"') {
+ return
+ }
+ }
+ }
+
+ private def isWhitespace(c: Int): Boolean = {
+ c == ' ' || c == '\t' || c == '\n' || c == '\r'
+ }
+
+ /** Returns the next character that is not JSON whitespace, or -1 at EOF. */
+ private def nextNonWhitespace(): Int = {
+ var c = nextChar()
+ while (isWhitespace(c)) {
+ c = nextChar()
+ }
+ c
+ }
+
+ /** Returns the next character and consumes it, or -1 at EOF. */
+ private def nextChar(): Int = {
+ if (pos >= bufferLen && !fill()) {
+ -1
+ } else {
+ val c = buffer(pos)
+ pos += 1
+ c
+ }
+ }
+
+ /** Returns the next character without consuming it, or -1 at EOF. */
+ private def peekChar(): Int = {
+ if (pos >= bufferLen && !fill()) {
+ -1
+ } else {
+ buffer(pos)
+ }
+ }
+
+ private def fill(): Boolean = {
+ pos = 0
+ bufferLen = reader.read(buffer)
+ bufferLen > 0
+ }
+}
+
+object EmbeddedArraySplitter {
+ private val BUFFER_SIZE = 64 * 1024
+}
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 8c69a4af2826..483c3c1f47f9 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
@@ -100,7 +100,7 @@ abstract class JsonDataSource extends Serializable with
Logging {
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions,
supportsArchiveScan: Boolean): Option[StructType] = {
- parsedOptions.singleVariantColumn match {
+
parsedOptions.singleVariantColumn.orElse(parsedOptions.explodeEmbeddedArray)
match {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
val hasArchive = parsedOptions.archiveFormatEnabled &&
@@ -214,7 +214,9 @@ abstract class JsonDataSource extends Serializable with
Logging {
object JsonDataSource {
def apply(options: JSONOptions): JsonDataSource = {
- if (options.multiLine) {
+ if (options.explodeEmbeddedArray.isDefined) {
+ EmbeddedArrayJsonDataSource
+ } else if (options.multiLine) {
MultiLineJsonDataSource
} else {
TextInputJsonDataSource
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/ExplodeEmbeddedArrayJsonSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/ExplodeEmbeddedArrayJsonSuite.scala
new file mode 100644
index 000000000000..075f61042d90
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/ExplodeEmbeddedArrayJsonSuite.scala
@@ -0,0 +1,396 @@
+/*
+ * 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.execution.datasources.json
+
+import java.io.{File, FileOutputStream, StringReader}
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.zip.GZIPOutputStream
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.Utils
+
+class EmbeddedArraySplitterSuite extends SparkFunSuite {
+ private def split(input: String, arrayFieldName: String = "Records"):
Seq[String] = {
+ val splitter = new EmbeddedArraySplitter(new StringReader(input),
arrayFieldName)
+
Iterator.continually(splitter.nextRecord()).takeWhile(_.isDefined).map(_.get).toSeq
+ }
+
+ test("basic record splitting") {
+ assert(split("""{"Records": [{"a": 1}, {"b": 2}]}""") === Seq("""{"a":
1}""", """{"b": 2}"""))
+ assert(split("""{"Records":[{"a":1}]}""") === Seq("""{"a":1}"""))
+ assert(split("""{"Records": []}""") === Seq.empty)
+ }
+
+ test("records with nesting, strings, and newlines") {
+ val record1 =
+ """{
+ | "a": {"nested": [1, 2, {"deep": "}]ignore me[{"}]},
+ | "b": "quote \" and backslash \\ and comma , and bracket ]"
+ |}""".stripMargin
+ val record2 = """{"Records": "a record field can also be called
Records"}"""
+ val input = s"""{"Records": [$record1,\n$record2\n]}"""
+ assert(split(input) === Seq(record1, record2))
+ }
+
+ test("non-record content is ignored") {
+ val input =
+ """{
+ | "before": {"Records": [{"not": "a real record"}]},
+ | "alsoBefore": [1, [2, {"x": "]"}]],
+ | "Records": [{"a": 1}],
+ | "after": "ignored"
+ |}""".stripMargin
+ assert(split(input) === Seq("""{"a": 1}"""))
+ }
+
+ test("scalar records") {
+ assert(split("""{"Records": [1, true, null, "s", 2.5]}""") ===
+ Seq("1", "true", "null", "\"s\"", "2.5"))
+ }
+
+ test("no matching array field yields nothing") {
+ assert(split("") === Seq.empty)
+ assert(split(" ") === Seq.empty)
+ assert(split("""{"a": 1}""") === Seq.empty)
+ // The field value is not an array.
+ assert(split("""{"Records": {"a": 1}}""") === Seq.empty)
+ // The top-level value is not an object (e.g. ndjson input).
+ assert(split("""[{"a": 1}]""") === Seq.empty)
+ assert(split("{\"a\": 1}\n{\"b\": 2}") === Seq.empty)
+ }
+
+ test("custom array field name") {
+ val input = """{"Records": [{"a": 1}], "events": [{"b": 2}, {"c": 3}]}"""
+ assert(split(input, "events") === Seq("""{"b": 2}""", """{"c": 3}"""))
+ assert(split(input) === Seq("""{"a": 1}"""))
+ // The field name is matched case-sensitively and by raw text: a key
written with escape
+ // sequences in the document does not match.
+ assert(split(input, "EVENTS") === Seq.empty)
+ assert(split("{\"ev\\u0065nts\": [{\"b\": 2}]}", "events") === Seq.empty)
+ }
+
+ test("byte order mark is skipped") {
+ val bom = 0xFEFF.toChar.toString // byte order mark (U+FEFF)
+ assert(split(bom + """{"Records": [{"a": 1}]}""") === Seq("""{"a": 1}"""))
+ }
+
+ test("truncated input returns the partial record") {
+ assert(split("""{"Records": [{"a": 1}, {"b": """) === Seq("""{"a": 1}""",
"""{"b": """))
+ assert(split("""{"Records": [{"a": "unclosed """) === Seq("""{"a":
"unclosed """))
+ // Truncated before any record.
+ assert(split("""{"Records": """) === Seq.empty)
+ assert(split("""{"Records"""") === Seq.empty)
+ }
+
+ test("input larger than the internal buffer") {
+ val records = (0 until 10000).map { i =>
+ s"""{"id": $i, "payload": "${"x" * (i % 97)}", "nested": {"level": [$i,
${i + 1}]}}"""
+ }
+ val input = s"""{"head": "ignored", "Records": [${records.mkString(",
")}], "tail": 1}"""
+ assert(input.length > 64 * 1024)
+ assert(split(input) === records)
+ }
+}
+
+abstract class ExplodeEmbeddedArrayJsonSuite extends QueryTest with
SharedSparkSession {
+ private val documentContent =
+ """{
+ | "unrelatedField": {"Records": [{"fake": 1}]},
+ | "Records": [
+ | {"eventName": "GetObject",
+ | "requestParameters": {"bucketName": "b1"}},
+ | {"eventName": "PutObject", "requestParameters": null},
+ | "scalar record",
+ | [1, 2]
+ | ],
+ | "trailing": "ignored"
+ |}""".stripMargin
+
+ private val expectedRecords = Seq(
+
Row("""{"eventName":"GetObject","requestParameters":{"bucketName":"b1"}}"""),
+ Row("""{"eventName":"PutObject","requestParameters":null}"""),
+ Row("\"scalar record\""),
+ Row("[1,2]"))
+
+ private def writeFile(dir: File, name: String, content: String): File = {
+ val file = new File(dir, name)
+ Files.write(file.toPath, content.getBytes(StandardCharsets.UTF_8))
+ file
+ }
+
+ protected def writeGzipFile(dir: File, name: String, content: String): File
= {
+ val file = new File(dir, name)
+ val out = new GZIPOutputStream(new FileOutputStream(file))
+ try {
+ out.write(content.getBytes(StandardCharsets.UTF_8))
+ } finally {
+ out.close()
+ }
+ file
+ }
+
+ test("basic embedded array scan") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", documentContent)
+ val df = spark.read.format("json").option("explodeEmbeddedArray",
"Records")
+ .load(dir.getAbsolutePath)
+ // The inferred schema names the variant column after the embedded array
field.
+ assert(df.schema === StructType.fromDDL("Records variant"))
+ checkAnswer(df.selectExpr("to_json(Records)"), expectedRecords)
+ assert(df.count() === 4)
+ }
+ }
+
+ test("array field name other than Records") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json",
+ """{"Records": [{"ignored": 1}], "events": [{"a": 1}, {"a": 2}]}""")
+ val df = spark.read.format("json").option("explodeEmbeddedArray",
"events")
+ .load(dir.getAbsolutePath)
+ assert(df.schema === StructType.fromDDL("events variant"))
+ checkAnswer(df.selectExpr("to_json(events)"), Seq(Row("""{"a":1}"""),
Row("""{"a":2}""")))
+ // A user-specified schema may name the variant column differently from
the array field.
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "events")
+ .schema("record variant")
+ .load(dir.getAbsolutePath).selectExpr("to_json(record)"),
+ Seq(Row("""{"a":1}"""), Row("""{"a":2}""")))
+ }
+ }
+
+ test("equivalent to reading the records as ndjson with singleVariantColumn")
{
+ withTempDir { dir =>
+ val records = (0 until 1000).map { i =>
+ s"""{"id": $i, "name": "event-$i", "values": [$i, {"nested":
"str,[\\"$i"}]}"""
+ }
+ val embeddedDir = new File(dir, "embedded")
+ val ndjsonDir = new File(dir, "ndjson")
+ Utils.createDirectory(embeddedDir)
+ Utils.createDirectory(ndjsonDir)
+ // Records in the array are separated by newlines, and some records span
multiple lines.
+ val arrayBody = records.map(_.replace(", ", ",\n ")).mkString(",\n")
+ writeFile(embeddedDir, "file.json", s"""{"Records": [\n$arrayBody\n]}""")
+ writeFile(ndjsonDir, "file.json", records.mkString("\n"))
+
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .load(embeddedDir.getAbsolutePath).selectExpr("to_json(Records)"),
+ spark.read.format("json").option("singleVariantColumn", "record")
+
.load(ndjsonDir.getAbsolutePath).selectExpr("to_json(record)").collect().toSeq)
+ }
+ }
+
+ test("malformed records") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", """{"Records": [{"a": 1}, 12ab, {"b":
2}]}""")
+
+ // PERMISSIVE (default) with a corrupt record column.
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .schema("record variant, _corrupt_record string")
+ .load(dir.getAbsolutePath)
+ .selectExpr("to_json(record)", "_corrupt_record"),
+ Seq(Row("""{"a":1}""", null), Row(null, "12ab"), Row("""{"b":2}""",
null)))
+
+ // DROPMALFORMED drops the malformed record.
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .option("mode", "DROPMALFORMED")
+ .load(dir.getAbsolutePath).selectExpr("to_json(Records)"),
+ Seq(Row("""{"a":1}"""), Row("""{"b":2}""")))
+
+ // FAILFAST fails the query.
+ val e = intercept[Exception] {
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .option("mode", "FAILFAST")
+ .load(dir.getAbsolutePath).collect()
+ }
+ assert(Utils.exceptionString(e).contains("MALFORMED_RECORD_IN_PARSING"))
+ }
+ }
+
+ test("truncated file") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", """{"Records": [{"a": 1}, {"b": """)
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .schema("record variant, _corrupt_record string")
+ .load(dir.getAbsolutePath)
+ .selectExpr("to_json(record)", "_corrupt_record"),
+ Seq(Row("""{"a":1}""", null), Row(null, """{"b": """)))
+ }
+ }
+
+ test("files without a matching array field yield no rows") {
+ withTempDir { dir =>
+ writeFile(dir, "empty.json", "")
+ writeFile(dir, "no_records.json", """{"a": 1}""")
+ writeFile(dir, "ndjson.json", "{\"a\": 1}\n{\"b\": 2}")
+ writeFile(dir, "empty_records.json", """{"Records": []}""")
+ writeFile(dir, "other_array.json", """{"events": [{"a": 1}]}""")
+ assert(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .load(dir.getAbsolutePath).count() === 0)
+ }
+ }
+
+ test("compressed file") {
+ withTempDir { dir =>
+ writeGzipFile(dir, "file.json.gz", documentContent)
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .load(dir.getAbsolutePath).selectExpr("to_json(Records)"),
+ expectedRecords)
+ }
+ }
+
+ test("multiple files and partition columns") {
+ withTempDir { dir =>
+ Utils.createDirectory(new File(dir, "p=1"))
+ Utils.createDirectory(new File(dir, "p=2"))
+ writeFile(dir, "p=1/file.json", """{"Records": [{"a": 1}, {"a": 2}]}""")
+ writeFile(dir, "p=2/file.json", """{"Records": [{"a": 3}]}""")
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .load(dir.getAbsolutePath).selectExpr("p", "to_json(Records)"),
+ Seq(Row(1, """{"a":1}"""), Row(1, """{"a":2}"""), Row(2,
"""{"a":3}""")))
+ }
+ }
+
+ test("multiLine option does not interfere") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", documentContent)
+ checkAnswer(
+ spark.read.format("json").option("explodeEmbeddedArray", "Records")
+ .option("multiLine", "true")
+ .load(dir.getAbsolutePath).selectExpr("to_json(Records)"),
+ expectedRecords)
+ }
+ }
+
+ test("streaming read") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", documentContent)
+ withTempDir { streamDir =>
+ val stream = spark.readStream.format("json")
+ .option("explodeEmbeddedArray", "Records")
+ .load(dir.getCanonicalPath)
+ .selectExpr("to_json(Records) as record")
+ .writeStream
+ .option("checkpointLocation", streamDir.getCanonicalPath +
"/checkpoint")
+ .format("parquet")
+ .start(streamDir.getCanonicalPath + "/output")
+ try {
+ stream.processAllAvailable()
+ } finally {
+ stream.stop()
+ }
+ checkAnswer(
+ spark.read.format("parquet").load(streamDir.getCanonicalPath +
"/output"),
+ expectedRecords)
+ }
+ }
+ }
+
+ test("user-specified schema constraints") {
+ withTempDir { dir =>
+ val file = writeFile(dir, "file.json", """{"Records": [{"a": 1}]}""")
+
+ // These are valid. The variant column may have any name.
+ for ((schema, options) <- Seq(
+ ("record variant", Map[String, String]()),
+ ("other variant", Map[String, String]()),
+ ("record variant, _corrupt_record string", Map[String, String]()),
+ ("c string, record variant", Map("columnNameOfCorruptRecord" -> "c"))
+ )) {
+ spark.read.options(Map("explodeEmbeddedArray" -> "Records") ++ options)
+ .schema(schema).json(file.getAbsolutePath).collect()
+ }
+
+ // These are invalid.
+ for ((schema, options) <- Seq(
+ ("record int", Map[String, String]()),
+ ("_corrupt_record variant", Map[String, String]()),
+ ("a variant, b variant", Map[String, String]()),
+ ("record variant, c string", Map[String, String]()),
+ ("record variant, _corrupt_record string, c string", Map[String,
String]())
+ )) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.options(Map("explodeEmbeddedArray" -> "Records") ++
options)
+ .schema(schema).json(file.getAbsolutePath).collect()
+ },
+ condition = "INVALID_EXPLODE_EMBEDDED_ARRAY_SCHEMA",
+ parameters = Map("schema" -> ("\"" + StructType.fromDDL(schema).sql
+ "\"")))
+ }
+ }
+ }
+
+ test("conflicting options") {
+ withTempDir { dir =>
+ writeFile(dir, "file.json", """{"Records": [{"a": 1}]}""")
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.format("json")
+ .option("explodeEmbeddedArray", "Records")
+ .option("singleVariantColumn", "var")
+ .load(dir.getAbsolutePath).collect()
+ },
+ condition = "EXPLODE_EMBEDDED_ARRAY_CONFLICTING_OPTION",
+ parameters = Map("option" -> "singleVariantColumn"))
+ }
+ }
+
+ test("rejected in from_json and json(Dataset[String])") {
+ import testImplicits._
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.option("explodeEmbeddedArray", "Records")
+ .json(Seq("""{"Records": [{"a": 1}]}""").toDS())
+ },
+ condition = "EXPLODE_EMBEDDED_ARRAY_UNSUPPORTED_USAGE",
+ parameters = Map("usage" -> "the json(Dataset[String]) API"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ """SELECT from_json('{"a": 1}', 'a INT', map('explodeEmbeddedArray',
'Records'))""")
+ .collect()
+ },
+ condition = "EXPLODE_EMBEDDED_ARRAY_UNSUPPORTED_USAGE",
+ parameters = Map("usage" -> "the from_json function"))
+ }
+}
+
+class ExplodeEmbeddedArrayJsonV1Suite extends ExplodeEmbeddedArrayJsonSuite {
+ override protected def sparkConf: SparkConf =
+ super
+ .sparkConf
+ .set(SQLConf.USE_V1_SOURCE_LIST, "json")
+}
+
+class ExplodeEmbeddedArrayJsonV2Suite extends ExplodeEmbeddedArrayJsonSuite {
+ override protected def sparkConf: SparkConf =
+ super
+ .sparkConf
+ .set(SQLConf.USE_V1_SOURCE_LIST, "")
+}
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 99be72b08a94..308b41d9ec77 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
@@ -3907,7 +3907,7 @@ abstract class JsonSuite
}
test("SPARK-40667: validate JSON Options") {
- assert(JSONOptions.getAllOptions.size == 32)
+ assert(JSONOptions.getAllOptions.size == 33)
// Please add validation on any new Json options here
assert(JSONOptions.isValidOption("samplingRatio"))
assert(JSONOptions.isValidOption("primitivesAsString"))
@@ -3938,6 +3938,7 @@ abstract class JsonSuite
assert(JSONOptions.isValidOption("timeZone"))
assert(JSONOptions.isValidOption("writeNonAsciiCharacterAsCodePoint"))
assert(JSONOptions.isValidOption("singleVariantColumn"))
+ assert(JSONOptions.isValidOption("explodeEmbeddedArray"))
assert(JSONOptions.isValidOption("useUnsafeRow"))
assert(JSONOptions.isValidOption("encoding"))
assert(JSONOptions.isValidOption("charset"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
