cloud-fan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4053512430
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,104 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = try {
+ createParser(factory, record)
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case e: CharConversionException if options.encoding.isEmpty =>
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ // Abandoning the iterator before END_ARRAY (e.g. a LIMIT) skips
finish()/fail(), so close the
+ // parser at task completion; close() is idempotent with those eager
closes.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
jsonParser.close()))
Review Comment:
**Non-blocking (P2):** The task-completion listener captures this JsonParser
directly, and closing the parser does not clear that captured reference. On
archive reads, the parser can retain the ByteArrayInputStream and complete byte
array for its entry, so consuming many entries retains all completed entry
buffers until task completion and can OOM despite the streaming path. Please
make the fallback cleanup stop retaining a parser once its eager close path has
run.
**Recommended change:** Replace the per-parser task-listener capture with
one clearable parser-lifecycle owner used by normal completion, terminal
failure, eager fallback, and task completion, and add lifecycle and
archive-entry retention regression coverage.
**Why this works:** The task listener captures only a clearable ownership
handle. Every eager close atomically consumes or clears that handle before
releasing the parser, while task completion consumes the same handle and closes
it only when an iterator was abandoned. Completed parsers therefore cease to be
reachable from TaskContext without weakening abandonment cleanup.
**Scope:** Unify streaming JsonParser close ownership across eager and
task-completion paths and verify that archive entries do not accumulate
completed parser state.
**Compatibility:** Streaming remains lazy, terminal and recoverable failures
retain their current parse-mode semantics, and task completion remains a
cleanup backstop for early iterator abandonment.
**Risks:** A close route could clear ownership before the parser is actually
closed, weakening cleanup after a close failure. Independent close paths could
race or mask the original parse failure if they do not consume the shared owner
consistently.
**Constraints:** Normal completion and terminal parse failures must still
close eagerly. Task completion must still close the parser when a consumer
abandons the iterator. Parse-mode recovery and lazy element delivery must
remain unchanged.
**Success:** After an entry iterator finishes or fails, TaskContext no
longer retains its parser or backing entry bytes. An iterator abandoned before
completion still has its current parser closed at task completion. Successful,
recoverable, and terminal parsing retain their current rows and parse-mode
behavior.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,104 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
Review Comment:
**Non-blocking (P2):** JacksonParser construction has already called
makeRootConverter(schema), whose struct branch builds the equivalent element
converter. Calling makeConverter(schema) again here duplicates recursive
converter construction for every file or archive entry, and the work is wasted
entirely for non-array roots. Please reuse the existing converter graph or
defer this construction until an array root has actually been observed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7034,6 +7034,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY =
+ buildConf("spark.sql.json.enableStreamingTopLevelArray")
+ .internal()
+ .doc("When true, multiline JSON reads stream the elements of a top-level
array one at a " +
Review Comment:
**Nit (P3):** This description makes the setting sound unconditional, but
the parser only selects the streaming path when neither singleVariantColumn nor
explodeEmbeddedArray is active. With either option, the top-level array is
still converted eagerly. Please qualify the description so it states the actual
option scope.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:
##########
@@ -1066,6 +1070,282 @@ abstract class JsonSuite
}
}
+ gridTest("SPARK-3308 Read multiline top level JSON arrays")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ withTempPath { file =>
+ Files.write(file.toPath,
"""[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8))
Review Comment:
**Non-blocking (P2):** The new enabled-path cases do not exercise either
zero-row branch: an empty document closes before the streaming iterator is
constructed, while an empty array closes from the iterator's first prepare
call. They also do not assert synchronous closure after normal array
exhaustion. Please add cases that assert both zero returned rows and parser
closure for the empty inputs, plus closure after fully consuming a normal array.
--
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]
