ganeshashree commented on code in PR #57559:
URL: https://github.com/apache/spark/pull/57559#discussion_r3705605942
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala:
##########
@@ -350,6 +357,462 @@ case class JsonTupleEvaluator(foldableFieldNames:
Array[Option[String]]) {
}
}
+/**
+ * The three-state result of navigating a JSON path for `JSON_TABLE`.
`get_json_object` collapses
+ * "the path is absent" and "the value is JSON null" into a single `null`,
which is wrong for
+ * `JSON_TABLE`: `EXISTS` must treat a present-but-null value as existing, and
a value column must
+ * distinguish SQL `NULL` from the literal string `"null"`. This ADT keeps the
two cases distinct.
+ */
+sealed trait JsonPathResult
+object JsonPathResult {
+ /** The path did not match (the key/index is absent). */
+ case object Missing extends JsonPathResult
+ /** The path matched a JSON `null` literal. */
+ case object NullValue extends JsonPathResult
+ /** The path matched a value; `raw` is its verbatim JSON text (including
quoted strings). */
+ case class Found(raw: UTF8String) extends JsonPathResult
+}
+
+/**
+ * A prefix trie over the (wildcard-free) column paths of a single
`JSON_TABLE` invocation, built
+ * once via [[JsonTableEvaluator.buildPathTrie]] and reused for every row. It
lets
+ * [[JsonTableEvaluator.navigateAll]] resolve all columns in a single
traversal of a row item
+ * instead of re-parsing the item once per column.
+ *
+ * Each node groups the paths that share a common prefix: `named`/`indexed`
hold the object-key and
+ * array-index steps to child nodes, and `terminals` lists the result-slot
indices of the columns
+ * whose path ends exactly at this node.
+ */
+private[expressions] final class JsonTablePathTrie {
+ // Result-slot indices of columns whose path terminates at this node.
+ var terminals: List[Int] = Nil
+ // Object-key children, keyed by field name.
+ val named: mutable.HashMap[String, JsonTablePathTrie] = mutable.HashMap.empty
+ // Array-index children, keyed by index.
+ val indexed: mutable.HashMap[Long, JsonTablePathTrie] = mutable.HashMap.empty
+
+ def hasChildren: Boolean = named.nonEmpty || indexed.nonEmpty
+
+ /** True if no column path was inserted (e.g. an ordinality-only table):
nothing to resolve. */
+ def isEmpty: Boolean = terminals.isEmpty && !hasChildren
+}
+
+/**
+ * The result of positioning a parser at a JSON path for the `JSON_TABLE` row
source (see
+ * `positionAt`). Like [[JsonPathResult]] it distinguishes a missing path from
a JSON `null`, but
+ * `AtValue` leaves the parser on the matched value's first token (rather than
serializing it) so
+ * the row source can be streamed.
+ */
+sealed trait PositionResult
+object PositionResult {
+ /** The path did not match. */
+ case object Missing extends PositionResult
+ /** The path matched a JSON `null` literal. */
+ case object NullValue extends PositionResult
+ /** The path matched a value; the parser is positioned at its first token. */
+ case object AtValue extends PositionResult
+}
+
+/**
+ * Row-source and column extraction for the SQL `JSON_TABLE` function. Given
the JSON input, the
+ * (wildcard-free) container path, and whether the row path ended in `[*]`, it
produces the
+ * per-row JSON documents that the
[[org.apache.spark.sql.catalyst.expressions.JsonTable]]
+ * generator then projects into columns via [[navigate]].
+ *
+ * - `$.items[*]` (containerPath `$.items`, `explodeRoot` = true): the
container must be an
+ * array; each element becomes a row.
+ * - `$` or `$.x` (`explodeRoot` = false): the matched value becomes exactly
one row.
+ *
+ * Unlike `get_json_object`, navigation here is token-aware and distinguishes
missing keys from
+ * JSON `null` values (see [[JsonPathResult]]).
+ *
+ * The input is required to be exactly one well-formed JSON value (no trailing
garbage, not
+ * empty); anything else is treated as malformed, so the caller applies the ON
ERROR behavior
+ * consistently in both modes.
+ */
+case class JsonTableEvaluator(containerPath: Seq[PathInstruction],
explodeRoot: Boolean) {
+ import PathInstruction._
+ import SharedFactory._
+
+ /**
+ * Returns the per-row JSON documents selected by the row path as an
iterator, or `None` if the
+ * JSON is null or malformed, or if `[*]` was applied to a non-array (the
caller maps `None` to
+ * the configured ON ERROR behavior). A well-formed input whose row path
matches nothing returns
+ * `Some(empty iterator)`.
+ *
+ * The input is first scanned once to validate it is a single well-formed
JSON value (so trailing
+ * garbage is rejected consistently in both ON ERROR modes -- this pass is
O(n) tokens and does
+ * not materialize values). For the array (`[*]`) case the elements are then
serialized one at a
+ * time from a second parser, so the whole expanded payload is never held in
memory at once.
+ */
+ final def evaluate(json: UTF8String): Option[Iterator[UTF8String]] = {
+ if (json == null || !isSingleWellFormedValue(json)) return None
+ // The parser is positioned at the matched value and, for the array case,
handed to a lazy
+ // iterator that reads elements directly from it -- the container is never
serialized whole.
+ // Ownership of `parser` transfers to that iterator (which closes it on
exhaustion); in every
+ // other branch we close it before returning.
+ val parser = CreateJacksonParser.utf8String(jsonFactory, json)
+ var transferred = false
+ try {
+ parser.nextToken()
+ positionAt(parser, containerPath) match {
+ case PositionResult.Missing =>
+ // Well-formed JSON, but the row path matched nothing: no rows.
+ Some(Iterator.empty)
+ case PositionResult.NullValue =>
+ // The container is JSON null. `[*]` over a non-array is an error;
otherwise one row.
+ if (explodeRoot) None else
Some(Iterator.single(UTF8String.fromString("null")))
+ case PositionResult.AtValue =>
+ if (explodeRoot) {
+ // `[*]` requires an array; a non-array match is an error.
+ if (parser.currentToken != JsonToken.START_ARRAY) {
+ None
+ } else {
+ val it = arrayElementIterator(parser) // owns and eventually
closes `parser`
+ transferred = true
+ Some(it)
+ }
+ } else {
+ Some(Iterator.single(serializeCurrentValue(parser)))
+ }
+ }
+ } catch {
+ case _: JsonProcessingException => None
+ } finally {
+ if (!transferred) parser.close()
+ }
+ }
+
+ /**
+ * Navigates `path` and leaves the parser positioned at the first token of
the matched value
+ * (returning `AtValue`), or returns `Missing`/`NullValue`. Unlike
[[navigateTo]] this does not
Review Comment:
Fixed. The contrast is now with the column projection traversal, linked as
`[[navigateColumns]]` (there's no separate serializing predecessor method), so
the reference resolves.
--
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]
