cloud-fan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4105396991
##########
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:
**Nit (P3):** `columnNameOfCorruptRecord` is the option that names the
destination field, not the field itself. Please say that PERMISSIVE fills the
field configured by `columnNameOfCorruptRecord`; the rest of the element-level
explanation can stay 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:
**Non-blocking (P2):** This new iterator route has no enabled case with an
explicit non-UTF-8 encoding, and the archive route has the same independent
wiring. Replacing either `streamParser` argument with the default parser would
leave the UTF-8 streaming tests and eager encoding tests green. Please cover an
explicitly encoded non-UTF-8 top-level array through both directory and archive
reads.
##########
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:
**Non-blocking (P2):** The compatibility boundary here is the behavior when
callers leave both settings unset, but every malformed-array test currently
forces this key to true or false. A future default flip would change ordinary
reads from document-level to element-level recovery without failing those
grids. Please add a malformed multiline top-level-array case with neither the
session key nor per-read option set and assert the eager whole-document result.
--
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]
