cloud-fan commented on code in PR #57559:
URL: https://github.com/apache/spark/pull/57559#discussion_r3709954204
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -3255,6 +3257,91 @@ class AstBuilder extends DataTypeAstBuilder
mayApplyAliasPlan(unnest.tableAlias, generate)
}
+ /**
+ * Create a plan for the SQL:2016 `JSON_TABLE` table-valued function. This
builds a
+ * [[Generate]] over the [[JsonTable]] generator (reusing the existing
Generate operator), so a
+ * downstream `SELECT` sees one output column per COLUMNS entry.
+ */
+ override def visitJsonTableRelation(
+ ctx: JsonTableRelationContext): LogicalPlan = withOrigin(ctx) {
+ val jt = ctx.jsonTable
+ val jsonExpr = expression(jt.jsonExpr)
+ val rowPath = string(visitStringLit(jt.rowPath))
+
+ val columns = jt.jsonTableColumn.asScala.map(buildJsonTableColumn).toSeq
+ // Column names must be unique within a single JSON_TABLE.
+ val duplicate =
columns.groupBy(_.name.toLowerCase(Locale.ROOT)).collectFirst {
Review Comment:
Duplicate-name validation needs to honor `spark.sql.caseSensitive`. This
unconditional folding rejects quoted `a` and `A` even when Spark's configured
resolver treats them as distinct. Please mirror the conditional normalization
used by the other parser duplicate checks and cover both modes.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala:
##########
@@ -350,6 +357,473 @@ 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 [[navigateColumns]].
+ *
+ * - `$.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 the
column projection traversal
+ * ([[navigateColumns]]), this does not serialize the value or finish
consuming the enclosing
+ * containers -- the caller either streams from the current position (array
row source) or
+ * serializes the single matched value.
+ */
+ private def positionAt(parser: JsonParser, path: Seq[PathInstruction]):
PositionResult = {
+ path match {
+ case Nil =>
+ if (parser.currentToken == JsonToken.VALUE_NULL)
PositionResult.NullValue
+ else PositionResult.AtValue
+
+ case Key :: Named(name) :: rest =>
+ if (parser.currentToken != JsonToken.START_OBJECT) {
+ skipRest(parser)
+ PositionResult.Missing
+ } else {
+ var token = parser.nextToken()
+ while (token != null && token != JsonToken.END_OBJECT) {
+ if (parser.currentName == name) {
+ parser.nextToken() // move onto the value; stop here (first
match wins)
+ return positionAt(parser, rest)
+ }
+ parser.nextToken()
+ parser.skipChildren()
+ token = parser.nextToken()
+ }
+ PositionResult.Missing
+ }
+
+ case Subscript :: Index(index) :: rest =>
+ if (parser.currentToken != JsonToken.START_ARRAY) {
+ skipRest(parser)
+ PositionResult.Missing
+ } else {
+ var i = 0L
+ var token = parser.nextToken()
+ while (token != null && token != JsonToken.END_ARRAY) {
+ if (i == index) {
+ return positionAt(parser, rest)
+ }
+ parser.skipChildren()
+ i += 1
+ token = parser.nextToken()
+ }
+ PositionResult.Missing
+ }
+
+ case _ =>
+ // Should not happen: JSON_TABLE paths are validated to be simple and
wildcard-free.
+ skipRest(parser)
+ PositionResult.Missing
+ }
+ }
+
+ /**
+ * Returns true if the input is exactly one well-formed JSON value with no
trailing content, so a
+ * valid prefix followed by garbage, or an empty document, is treated as
malformed (consistently
+ * in both ON ERROR modes).
+ */
+ private def isSingleWellFormedValue(json: UTF8String): Boolean = {
+ try {
+ Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, json))
{ parser =>
+ if (parser.nextToken() == null) {
+ false // empty or whitespace-only
+ } else {
+ parser.skipChildren() // consume the first value in full
+ parser.nextToken() == null // nothing must remain after it
+ }
+ }
+ } catch {
+ case _: JsonProcessingException => false
+ }
+ }
+
+ /**
+ * Resolves every column of `trie` against the value at the parser's current
token in a single
+ * traversal, writing each matched terminal's [[JsonPathResult]] into `out`
at its slot index.
+ * Only the simple wildcard-free instruction set produced for `JSON_TABLE`
paths is modeled by the
+ * trie (`Key`/`Named` object steps and `Subscript`/`Index` array steps).
+ *
+ * Slots left untouched keep their initial `Missing`. A matched value is
stored as its raw JSON
+ * text (`Found.raw`), i.e. strings keep their enclosing quotes so the
fragment stays
+ * re-parseable; value columns unquote scalar strings afterwards via
[[JsonTable]]'s extraction.
+ */
+ private def navigateAll(
+ parser: JsonParser,
+ trie: JsonTablePathTrie,
+ out: Array[JsonPathResult]): Unit = {
+ val isNull = parser.currentToken == JsonToken.VALUE_NULL
+
+ if (!trie.hasChildren) {
+ // Leaf node: every column terminates here, so just record the current
value (or null) and
+ // consume it. This is the common case for disjoint column paths.
+ if (trie.terminals.nonEmpty) {
+ val result = if (isNull) JsonPathResult.NullValue
+ else JsonPathResult.Found(serializeCurrentValue(parser))
+ trie.terminals.foreach(out(_) = result)
+ } else {
+ skipRest(parser)
+ }
+ } else if (trie.terminals.nonEmpty && !isNull) {
+ // A column path both ends here and extends deeper (e.g. `$.a` alongside
`$.a.b`). Serialize
+ // the value once for the terminals, then re-parse that fragment to
resolve the deeper
+ // columns -- so the only place a value is parsed more than once is this
rare prefix overlap,
Review Comment:
Please narrow this claim to the `navigateAll` traversal. Array row items are
serialized by `arrayElementIterator` and then parsed again by
`navigateColumns`, so prefix overlap is not the only reparse in the overall
JSON_TABLE pipeline.
--
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]
