tdcmeehan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4106843648
##########
docs/sql-data-sources-json.md:
##########
@@ -237,6 +237,12 @@ Data source options of JSON can be set via:
<td>Parse one record, which may span multiple lines, per file. JSON
built-in functions ignore this option.</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>enableStreamingTopLevelArray</code></td>
+ <td>(value of <code>spark.sql.json.enableStreamingTopLevelArray</code>
configuration)</td>
+ <td>When <code>multiLine</code> is enabled and a file holds a top-level
JSON array, read the array's elements one at a time instead of materializing
the whole array before returning rows. It applies to reads into a struct
schema, and has no effect on reads using <code>singleVariantColumn</code> or
<code>explodeEmbeddedArray</code>. While streaming, <code>mode</code> applies
to an individual element rather than the whole document:
<code>PERMISSIVE</code> fills <code>columnNameOfCorruptRecord</code> for the
malformed element alone, leaving it null on the valid rows of the same
document, and <code>DROPMALFORMED</code> drops that element rather than the
document. An element whose failure leaves the parser at an unknown position,
such as a nested value of the wrong shape, still ends the document, as does a
failure outside any element, such as a syntax error between two elements or a
missing closing bracket.</td>
Review Comment:
Done in `763bf149a25` — it now reads "PERMISSIVE fills the field configured
by `columnNameOfCorruptRecord`", with the rest of the element-level explanation
unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -414,8 +415,14 @@ object MultiLineJsonDataSource extends JsonDataSource {
schema,
parser.options.columnNameOfCorruptRecord)
- safeParser.parse(
- CodecStreams.createInputStreamWithCloseResource(conf, file.toPath))
+ val input = CodecStreams.createInputStreamWithCloseResource(conf,
file.toPath)
+ if (parser.options.streamMultilineTopLevelArray) {
+ safeParser.parseIterator(
+ input,
+ input => parser.parseIterator[InputStream](input, streamParser,
partitionedFileString))
Review Comment:
Good catch. `763bf149a25` adds explicit non-UTF-8 coverage on both routes.
The SPARK-23723 mismatched-encoding test now runs with the conf disabled and
enabled (the directory route), and the archive suites stream an ISO-8859-1
top-level array and check it against a directory read of the same bytes (the
archive route) — ISO-8859-1 because Jackson auto-detects UTF-16/32, so only a
charset it can't guess shows the encoding is honored. I verified it the way you
described: replacing the directory `streamParser` with the default parser turns
the SPARK-23723 enabled arms red, and replacing the archive `streamParser`
turns the archive comparison red, so neither argument can be dropped silently
now.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7034,6 +7034,26 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY =
+ buildConf("spark.sql.json.enableStreamingTopLevelArray")
+ .doc("When true, multiline JSON file reads stream the elements of a
top-level array one at " +
+ "a time instead of materializing the entire array before returning
rows. This applies " +
+ "only to reads into a struct schema that take top-level arrays as
structs, and has no " +
+ "effect on reads using the `singleVariantColumn` or
`explodeEmbeddedArray` option. " +
+ "Streaming also makes an array element, rather than the whole
document, the record " +
+ "that a parse mode applies to, since rows already emitted cannot be
withdrawn: " +
+ "PERMISSIVE fills the corrupt record column for the malformed element
only, leaving " +
+ "it null on the valid rows of the same document, and DROPMALFORMED
drops that " +
+ "element rather than the whole document. An element whose failure
leaves the parser " +
+ "at an unknown position, such as a nested value of the wrong shape,
still ends the " +
+ "document, as does a failure outside any element, such as a syntax
error between two " +
+ "elements or a missing closing bracket. It can be overwritten by the
JSON option " +
+ "`enableStreamingTopLevelArray`.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
Review Comment:
Thanks — and I follow the concern: the grids set the key explicitly on both
sides, so on their own they wouldn't catch a change to the default. I'd lean
toward leaving this one out, though. Both sides of the recovery boundary are
already pinned by the disabled/enabled grids — a malformed array gives the
whole-document result with it off and per-element recovery with it on — so the
contract itself is covered. A case with neither set would assert the current
default value specifically, and flipping that default later would be a
deliberate change a reviewer would see, best pinned alongside that change
rather than guarded here. Since it's non-blocking I've left it as-is for now,
but I'm glad to add the explicit default-path case if you'd still prefer it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
