This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 3ef46172fdfe [SPARK-58366][SQL] Support JSON_TABLE table-valued
function
3ef46172fdfe is described below
commit 3ef46172fdfe3e077dd35262def89f699e7510bf
Author: Ganesha Shreedhara <[email protected]>
AuthorDate: Tue Aug 4 20:11:48 2026 +0800
[SPARK-58366][SQL] Support JSON_TABLE table-valued function
### What changes were proposed in this pull request?
Add the ANSI SQL:2016 `JSON_TABLE` table-valued function, which shreds a
JSON document into a relational table. A row path selects a sequence of JSON
items and a `COLUMNS` clause projects a typed value out of each item into a
column.
Syntax (flat, non-nested subset):
JSON_TABLE(json_expr, row_path
COLUMNS (
col1 FOR ORDINALITY,
col2 <type> [PATH '<json_path>'],
col3 <type> EXISTS [PATH '<json_path>']
)
[ { NULL | ERROR } ON ERROR ]
) [AS] alias
- A row path ending in `[*]` expands a JSON array into one row per element;
a non-wildcard path yields a single row for the matched value.
- `FOR ORDINALITY`: a 1-based BIGINT row counter.
- Value columns are extracted and cast to the declared type; the path may
be explicit (`PATH '...'`) or implicit (`'$.<columnName>'`).
- `EXISTS` columns are a presence test cast to the declared type; a
present-but-null JSON value counts as existing, only an absent path is false.
- `{ NULL | ERROR } ON ERROR`: `NULL ON ERROR` (default) yields no rows on
null/malformed input; `ERROR ON ERROR` raises.
- Usable in a comma join and with `LATERAL`.
Implementation notes:
- A `JsonTable` `Generator` expression is wrapped by the existing
`Generate` operator, so no new execution operator is introduced.
- A dedicated grammar production handles the `COLUMNS(...)` clause;
`AstBuilder.visitJsonTableRelation` builds the plan. `ERROR`, `ORDINALITY`, and
`JSON_TABLE` are added as non-reserved keywords.
- A token-aware navigator (`JsonTableEvaluator`) extracts values so that a
missing path, a JSON `null`, and the literal string `"null"` are all
distinguished (unlike `get_json_object`). Non-trailing wildcard paths are
rejected during analysis via `DATATYPE_MISMATCH.INVALID_JSON_TABLE_PATH`.
- Only the flat subset is implemented; `NESTED PATH`, `FORMAT JSON` query
columns, and `DEFAULT ... ON EMPTY` are left as follow-ups.
### Why are the changes needed?
`JSON_TABLE` is the SQL-standard way to turn JSON into rows and columns,
and is supported by Oracle, DB2, MySQL 8, PostgreSQL 17, Snowflake, and Trino.
Spark previously required chaining `from_json` + `explode`/`inline` +
`get_json_object` to achieve the same result. This folds that into one
declarative, standard construct and eases migration from those systems.
### Does this PR introduce _any_ user-facing change?
Yes. It adds the `JSON_TABLE` SQL table-valued function and a SQL reference
documentation page. `ERROR`, `ORDINALITY`, and `JSON_TABLE` are added as
non-reserved keywords, so existing queries using them as identifiers continue
to parse. There is no change to existing behavior.
### How was this patch tested?
New end-to-end suite `JsonTableSuite` covering array expansion, ordinality,
explicit/implicit paths, missing-vs-null semantics for value and EXISTS
columns, `NULL`/`ERROR ON ERROR`, `[*]` over a non-array, mid-path wildcard
rejection, non-explode string row items, arrays of scalars/strings, duplicate
keys, structure-valued columns, `LATERAL` joins, nested-field extraction, and
column aliases. Existing `JsonExpressionsSuite`, `JsonFunctionsSuite`,
`GeneratorFunctionSuite`, `PlanParserS [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57559 from ganeshashree/SPARK-58366.
Lead-authored-by: Ganesha Shreedhara <[email protected]>
Co-authored-by: Ganesha S <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit ce0578d679d52f0399e0b5100b484cde072316b2)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 10 +
docs/sql-ref-ansi-compliance.md | 2 +
docs/sql-ref-syntax-qry-select-json-table.md | 130 ++++
docs/sql-ref-syntax-qry-select.md | 1 +
docs/sql-ref-syntax.md | 1 +
.../spark/sql/catalyst/parser/SqlBaseLexer.g4 | 2 +
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 26 +
.../spark/sql/errors/QueryParsingErrors.scala | 7 +
.../expressions/json/JsonExpressionEvalUtils.scala | 484 ++++++++++++-
.../sql/catalyst/expressions/jsonExpressions.scala | 294 +++++++-
.../spark/sql/catalyst/parser/AstBuilder.scala | 91 +++
.../jdbc/SparkConnectDatabaseMetaDataSuite.scala | 2 +-
.../sql-tests/results/keywords-enforced.sql.out | 2 +
.../resources/sql-tests/results/keywords.sql.out | 2 +
.../sql-tests/results/nonansi/keywords.sql.out | 2 +
.../org/apache/spark/sql/JsonTableSuite.scala | 800 +++++++++++++++++++++
.../ThriftServerWithSparkContextSuite.scala | 2 +-
17 files changed, 1849 insertions(+), 9 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 23d7f69a5485..69dfa0cc68cb 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1941,6 +1941,11 @@
"Input schema <schema> must be a struct, an array, a map or a
variant."
]
},
+ "INVALID_JSON_TABLE_PATH" : {
+ "message" : [
+ "The <location> of JSON_TABLE has an invalid or unsupported JSON
path <path>. Only simple, wildcard-free paths are supported (a trailing '[*]'
is allowed on the row path)."
+ ]
+ },
"INVALID_MAP_KEY_TYPE" : {
"message" : [
"The key of map cannot be/contain <keyType>."
@@ -5032,6 +5037,11 @@
"CREATE TEMPORARY TABLE ... USING ... is a deprecated syntax. To
overcome the issue, please use CREATE TEMPORARY VIEW instead."
]
},
+ "DUPLICATE_JSON_TABLE_COLUMN" : {
+ "message" : [
+ "JSON_TABLE has duplicate column name <columnName>. Column names in
the COLUMNS clause must be unique."
+ ]
+ },
"EMPTY_IN_PREDICATE" : {
"message" : [
"IN predicate requires at least one value. Empty IN clauses like 'IN
()' are not allowed. Consider using 'WHERE FALSE' if you need an always-false
condition, or provide at least one value in the IN list."
diff --git a/docs/sql-ref-ansi-compliance.md b/docs/sql-ref-ansi-compliance.md
index 0b8c7c99b77f..4eb886a16e60 100644
--- a/docs/sql-ref-ansi-compliance.md
+++ b/docs/sql-ref-ansi-compliance.md
@@ -539,6 +539,7 @@ Below is a list of all the keywords in Spark SQL.
|ELSEIF|non-reserved|non-reserved|non-reserved|
|END|reserved|non-reserved|reserved|
|ENFORCED|non-reserved|non-reserved|non-reserved|
+|ERROR|non-reserved|non-reserved|non-reserved|
|ESCAPE|reserved|non-reserved|reserved|
|ESCAPED|non-reserved|non-reserved|non-reserved|
|EVOLUTION|non-reserved|non-reserved|non-reserved|
@@ -616,6 +617,7 @@ Below is a list of all the keywords in Spark SQL.
|ITERATE|non-reserved|non-reserved|non-reserved|
|JOIN|reserved|strict-non-reserved|reserved|
|JSON|non-reserved|non-reserved|non-reserved|
+|JSON_TABLE|non-reserved|non-reserved|reserved|
|KEY|non-reserved|non-reserved|non-reserved|
|KEYS|non-reserved|non-reserved|non-reserved|
|LANGUAGE|non-reserved|non-reserved|reserved|
diff --git a/docs/sql-ref-syntax-qry-select-json-table.md
b/docs/sql-ref-syntax-qry-select-json-table.md
new file mode 100644
index 000000000000..a8f40513778c
--- /dev/null
+++ b/docs/sql-ref-syntax-qry-select-json-table.md
@@ -0,0 +1,130 @@
+---
+layout: global
+title: JSON_TABLE
+displayTitle: JSON_TABLE
+license: |
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+---
+
+### Description
+
+The `JSON_TABLE` table-valued function shreds a JSON document into a
relational table. A
+_row path_ selects a sequence of JSON items, and a `COLUMNS` clause projects a
value out of each
+item into a typed column. This is the SQL-standard way (SQL:2016) to turn JSON
into rows and
+columns, and is commonly used to migrate queries from other systems such as
Oracle, DB2, and
+MySQL.
+
+Only the flat (non-nested) form is currently supported. `NESTED PATH` columns
are not yet
+supported.
+
+### Syntax
+
+```sql
+JSON_TABLE ( json_expr, row_path COLUMNS ( column_definition [ , ... ] ) [
error_clause ] ) [ table_alias ]
+
+column_definition
+ { column_name FOR ORDINALITY
+ | column_name data_type [ PATH json_path ]
+ | column_name data_type EXISTS [ PATH json_path ] }
+
+error_clause
+ { NULL | ERROR } ON ERROR
+```
+
+### Parameters
+
+* **json_expr**
+
+ An expression that evaluates to a `STRING` containing the JSON document.
+
+* **row_path**
+
+ A SQL/JSON path literal that selects the row source. A path ending in
`[*]` (for example
+ `'$.items[*]'`) selects each element of the matched array as a separate
row. Any other path
+ (for example `'$'`) selects a single value as one row. If the path matches
nothing, no rows
+ are produced.
+
+* **column_name FOR ORDINALITY**
+
+ Declares a `BIGINT` column that is a 1-based sequential counter of the
generated rows.
+
+* **column_name data_type [ PATH json_path ]**
+
+ A value column. The value at `json_path` (relative to a row item) is
extracted and cast to
+ `data_type`. If `PATH` is omitted, the path defaults to the column name
read as a single
+ object key: a simple identifier maps like `$.name`, while a name
containing special characters
+ such as a dot is treated as one literal key (for example a column named
`a.b` reads the key
+ `"a.b"`, equivalent to `$['a.b']`, not the nested path `a` -> `b`). If the
path matches nothing,
+ the column is `NULL`.
+
+* **column_name data_type EXISTS [ PATH json_path ]**
+
+ An existence column. Evaluates to a truthy value when `json_path` matches
and a falsy value
+ otherwise, cast to `data_type` (for example `BOOLEAN`).
+
+* **{ NULL | ERROR } ON ERROR**
+
+ Controls behavior when `json_expr` is `NULL` or not well-formed JSON.
`NULL ON ERROR` (the
+ default) produces no rows. `ERROR ON ERROR` raises an error.
+
+* **table_alias**
+
+ An optional alias for the output, optionally followed by a column alias
list.
+
+### Examples
+
+```sql
+-- Expand a JSON array into rows with typed columns and an ordinality counter
+SELECT t.* FROM JSON_TABLE(
+ '{"items":[{"id":1,"n":"a"},{"id":2,"n":"b"}]}',
+ '$.items[*]'
+ COLUMNS (
+ seq FOR ORDINALITY,
+ id INT PATH '$.id',
+ name STRING PATH '$.n'
+ )
+) AS t;
++---+---+----+
+|seq| id|name|
++---+---+----+
+| 1| 1| a|
+| 2| 2| b|
++---+---+----+
+
+-- Implicit column path derived from the column name, and an EXISTS column
+SELECT * FROM JSON_TABLE(
+ '{"rows":[{"id":10,"opt":1},{"id":20}]}',
+ '$.rows[*]'
+ COLUMNS (id INT, hasOpt BOOLEAN EXISTS PATH '$.opt')
+) AS t;
++---+------+
+| id|hasOpt|
++---+------+
+| 10| true|
+| 20| false|
++---+------+
+
+-- Join JSON_TABLE output against a base table using LATERAL
+SELECT d.id, t.k
+FROM docs d,
+LATERAL JSON_TABLE(d.doc, '$.tags[*]' COLUMNS (k STRING PATH '$.k')) AS t;
+```
+
+### Related Statements
+
+* [SELECT](sql-ref-syntax-qry-select.html)
+* [Table-valued Function](sql-ref-syntax-qry-select-tvf.html)
+* [LATERAL VIEW Clause](sql-ref-syntax-qry-select-lateral-view.html)
diff --git a/docs/sql-ref-syntax-qry-select.md
b/docs/sql-ref-syntax-qry-select.md
index a97ee766f216..d07f95e6b7b6 100644
--- a/docs/sql-ref-syntax-qry-select.md
+++ b/docs/sql-ref-syntax-qry-select.md
@@ -93,6 +93,7 @@ SELECT [ hints , ... ] [ ALL | DISTINCT ] { [ [
named_expression | regex_column_
* [Pivot relation](sql-ref-syntax-qry-select-pivot.html)
* [Unpivot relation](sql-ref-syntax-qry-select-unpivot.html)
* [Table-value function](sql-ref-syntax-qry-select-tvf.html)
+ * [JSON_TABLE](sql-ref-syntax-qry-select-json-table.html)
* [Inline table](sql-ref-syntax-qry-select-inline-table.html)
* [UNNEST relation](sql-ref-syntax-qry-select-unnest.html)
* [ [LATERAL](sql-ref-syntax-qry-select-lateral-subquery.html) ] (
Subquery )
diff --git a/docs/sql-ref-syntax.md b/docs/sql-ref-syntax.md
index 42e52bf04454..36cd297aaf82 100644
--- a/docs/sql-ref-syntax.md
+++ b/docs/sql-ref-syntax.md
@@ -83,6 +83,7 @@ ability to generate logical and physical plan for a given
query using
* [SORT BY Clause](sql-ref-syntax-qry-select-sortby.html)
* [TABLESAMPLE](sql-ref-syntax-qry-select-sampling.html)
* [Table-valued Function](sql-ref-syntax-qry-select-tvf.html)
+ * [JSON_TABLE](sql-ref-syntax-qry-select-json-table.html)
* [WHERE Clause](sql-ref-syntax-qry-select-where.html)
* [Aggregate Function](sql-ref-syntax-qry-select-aggregate.html)
* [Window Function](sql-ref-syntax-qry-select-window.html)
diff --git
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
index fea9d6ef7bb3..5e15b72d6a30 100644
---
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
+++
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
@@ -256,6 +256,7 @@ ELSE: 'ELSE';
ELSEIF: 'ELSEIF';
END: 'END';
ENFORCED: 'ENFORCED';
+ERROR: 'ERROR';
ESCAPE: 'ESCAPE';
ESCAPED: 'ESCAPED';
EVOLUTION: 'EVOLUTION';
@@ -333,6 +334,7 @@ ITEMS: 'ITEMS';
ITERATE: 'ITERATE';
JOIN: 'JOIN';
JSON: 'JSON';
+JSON_TABLE: 'JSON_TABLE';
KEY: 'KEY';
KEYS: 'KEYS';
LANGUAGE: 'LANGUAGE';
diff --git
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
index 87a253426edf..98fb1cd5f29f 100644
---
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
+++
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
@@ -1209,6 +1209,7 @@ relationPrimary
watermarkClause? tableAlias #aliasedRelation
| inlineTable
#inlineTableDefault2
| unnest #unnestTable
+ | jsonTable #jsonTableRelation
| tableFunctionCallWithTrailingClauses
#tableValuedFunction
;
@@ -1220,6 +1221,27 @@ unnest
(WITH ORDINALITY)? tableAlias
;
+// The ANSI SQL:2016 JSON_TABLE table-valued function. Because the COLUMNS
clause is not a normal
+// function-argument list, it has a dedicated production rather than going
through
+// tableFunctionCall. Only the flat (non-NESTED) subset is currently supported.
+jsonTable
+ : JSON_TABLE LEFT_PAREN jsonExpr=expression COMMA rowPath=stringLit
+ COLUMNS LEFT_PAREN jsonTableColumn (COMMA jsonTableColumn)* RIGHT_PAREN
+ jsonTableOnErrorClause?
+ RIGHT_PAREN tableAlias
+ ;
+
+jsonTableColumn
+ : colName=errorCapturingIdentifier FOR ORDINALITY
#jsonTableOrdinalityColumn
+ | colName=errorCapturingIdentifier dataType
+ EXISTS (PATH path=stringLit)?
#jsonTableExistsColumn
+ | colName=errorCapturingIdentifier dataType (PATH path=stringLit)?
#jsonTableValueColumn
+ ;
+
+jsonTableOnErrorClause
+ : (NULL | ERROR) ON ERROR
+ ;
+
optionsClause
: WITH options=propertyList
;
@@ -2150,6 +2172,7 @@ ansiNonReserved
| DROP
| ELSEIF
| ENFORCED
+ | ERROR
| ESCAPED
| EVOLUTION
| EXACT
@@ -2208,6 +2231,7 @@ ansiNonReserved
| ITEMS
| ITERATE
| JSON
+ | JSON_TABLE
| KEY
| KEYS
| LANGUAGE
@@ -2570,6 +2594,7 @@ nonReserved
| ELSEIF
| END
| ENFORCED
+ | ERROR
| ESCAPE
| ESCAPED
| EVOLUTION
@@ -2642,6 +2667,7 @@ nonReserved
| ITEMS
| ITERATE
| JSON
+ | JSON_TABLE
| KEY
| KEYS
| LANGUAGE
diff --git
a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
index 7cf205e75ac4..98af9876ccda 100644
---
a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
+++
b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
@@ -138,6 +138,13 @@ private[sql] object QueryParsingErrors extends
DataTypeErrorsBase {
ctx)
}
+ def duplicateJsonTableColumnError(columnName: String, ctx:
JsonTableContext): Throwable = {
+ new ParseException(
+ errorClass = "INVALID_SQL_SYNTAX.DUPLICATE_JSON_TABLE_COLUMN",
+ messageParameters = Map("columnName" -> toSQLId(columnName)),
+ ctx)
+ }
+
def clausesWithPipeOperatorsUnsupportedError(
ctx: QueryOrganizationContext,
clauses: String): Throwable = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
index e80b4a355b72..fbb81bccaa80 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
@@ -24,7 +24,7 @@ import scala.util.parsing.combinator.RegexParsers
import com.fasterxml.jackson.core._
import com.fasterxml.jackson.core.json.JsonReadFeature
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, TaskContext}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{ExprUtils,
GenericInternalRow, GetJsonObject}
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser,
JacksonGenerator, JacksonParser, JsonInferSchema, JSONOptions}
@@ -55,9 +55,9 @@ object JsonPathParser extends RegexParsers {
def root: Parser[Char] = '$'
- def long: Parser[Long] = "\\d+".r ^? {
- case x => x.toLong
- }
+ // Guard the conversion so an oversized index (e.g.
`[999999999999999999999999]`) makes the path
+ // fail to parse rather than throwing NumberFormatException out of the
parser.
+ def long: Parser[Long] = "\\d+".r ^? { case x if x.toLongOption.isDefined =>
x.toLong }
// parse `[*]` and `[123]` subscripts
def subscript: Parser[List[PathInstruction]] =
@@ -97,6 +97,13 @@ object JsonPathParser extends RegexParsers {
None
}
}
+
+ /**
+ * Returns `Some(true)` if the path parses and contains a wildcard,
`Some(false)` if it parses
+ * without a wildcard, and `None` if it does not parse. Used by `JSON_TABLE`
to validate that
+ * column and (container) row paths are simple, wildcard-free paths.
+ */
+ def hasWildcard(str: String): Option[Boolean] =
parse(str).map(_.contains(Wildcard))
}
private[this] object SharedFactory {
@@ -350,6 +357,475 @@ 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 -- this rare prefix overlap is the only place *within a
single traversal* that a
+ // value is parsed more than once, and even then the descendant columns
are still resolved in
+ // a single sub-traversal. (Separately, an array row item is serialized
by
+ // `arrayElementIterator` and parsed again here, once per row.)
+ val raw = serializeCurrentValue(parser)
+ val result = JsonPathResult.Found(raw)
+ trie.terminals.foreach(out(_) = result)
+ Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, raw))
{ sub =>
+ sub.nextToken()
+ descendInto(sub, trie, out)
+ }
+ } else {
+ // Descend for the deeper columns. Any terminal ending at this node
resolves to `NullValue`:
+ // the earlier `!isNull` branch already handled non-null terminals, so a
terminal reaching
+ // here means the value is a JSON null. Descendant-only slots that do
not match stay `Missing`
+ // (a null has no children, so `descendInto` skips it and leaves them
untouched).
+ if (trie.terminals.nonEmpty) trie.terminals.foreach(out(_) =
JsonPathResult.NullValue)
+ descendInto(parser, trie, out)
+ }
+ }
+
+ /**
+ * Descends into the object or array at the parser's current token,
dispatching each matching
+ * field/element to the corresponding child trie node via [[navigateAll]]
and skipping the rest.
+ * A scalar (or JSON null) has no children, so the whole value is skipped
and the deeper columns
+ * are left as `Missing`.
+ */
+ private def descendInto(
+ parser: JsonParser,
+ trie: JsonTablePathTrie,
+ out: Array[JsonPathResult]): Unit = {
+ parser.currentToken match {
+ case JsonToken.START_OBJECT if trie.named.isEmpty =>
+ // No object-key columns descend here (only array-index paths): skip
the whole object.
+ skipRest(parser)
+
+ case JsonToken.START_OBJECT =>
+ // First match wins for duplicate keys: once a trie key has been
dispatched, later fields
+ // with the same name are skipped.
+ val consumed = mutable.HashSet.empty[String]
+ var token = parser.nextToken()
+ while (token != null && token != JsonToken.END_OBJECT) {
+ val name = parser.currentName
+ val child = trie.named.get(name)
+ parser.nextToken() // move onto the field value
+ if (child.isDefined && consumed.add(name)) {
+ navigateAll(parser, child.get, out)
+ } else {
+ parser.skipChildren()
+ }
+ token = parser.nextToken()
+ }
+
+ case JsonToken.START_ARRAY if trie.indexed.isEmpty =>
+ // No array-index columns descend here (only object-key paths): skip
the whole array.
+ skipRest(parser)
+
+ case JsonToken.START_ARRAY =>
+ var i = 0L
+ var token = parser.nextToken()
+ while (token != null && token != JsonToken.END_ARRAY) {
+ val child = trie.indexed.get(i)
+ if (child.isDefined) {
+ navigateAll(parser, child.get, out)
+ } else {
+ parser.skipChildren()
+ }
+ i += 1
+ token = parser.nextToken()
+ }
+
+ case _ =>
+ // A scalar where some columns expected to descend: those stay Missing.
+ skipRest(parser)
+ }
+ }
+
+ /** Skips the remainder of the value at the parser's current token. */
+ private def skipRest(parser: JsonParser): Unit = parser.skipChildren()
+
+ /**
+ * Serializes the value at the parser's current token to its raw JSON text.
Strings keep their
+ * enclosing quotes, so the result is always a re-parseable JSON fragment
(this matters because a
+ * matched value may be re-parsed as a row item). Value columns unquote
scalar strings afterwards
+ * via [[JsonTableEvaluator.unquotedString]].
+ */
+ private def serializeCurrentValue(parser: JsonParser): UTF8String = {
+ val output = new ByteArrayOutputStream()
+ Utils.tryWithResource(jsonFactory.createGenerator(output,
JsonEncoding.UTF8)) {
+ // `copyCurrentStructureExact` preserves floating-point tokens
byte-for-byte; the plain
+ // `copyCurrentStructure` may round them for textual formats, which
would corrupt a
+ // high-precision fraction before JSON_TABLE casts the reserialized text
to DECIMAL/STRING.
+ generator => generator.copyCurrentStructureExact(parser)
+ }
+ UTF8String.fromBytes(output.toByteArray)
+ }
+
+ // The array parser currently owned by an outstanding
`arrayElementIterator`, or null. Since
+ // `GenerateExec` evaluates rows sequentially and fully drains each row's
iterator before the
+ // next `eval`, at most one such parser is open at a time per task. Tracked
so a single
+ // task-completion listener (registered once below) can close it on early
termination.
+ @transient private var openArrayParser: JsonParser = _
+ @transient private var completionListenerRegistered = false
+
+ /**
+ * Streams the elements of the array the `parser` is currently positioned at
(`START_ARRAY`),
+ * serializing one element at a time straight from the source parser -- the
enclosing array is
+ * never materialized as a whole. The iterator owns `parser`: it closes it
on exhaustion (the
+ * fast path). To also close it when the consumer stops early (e.g. a
downstream `LIMIT`, or a
+ * per-column cast failure) -- the `Generator` API has no close hook -- a
*single* task-completion
+ * listener is registered per evaluator (i.e. per task) that closes
whichever parser is currently
+ * open, rather than one listener per input row, so processing many JSON
rows does not accumulate
+ * an unbounded listener list. `Generate` can thus emit rows for a large
array without holding the
+ * full expanded payload in memory.
+ */
+ private def arrayElementIterator(parser: JsonParser): Iterator[UTF8String] =
{
+ openArrayParser = parser
+ if (!completionListenerRegistered) {
+ Option(TaskContext.get()).foreach { tc =>
+ tc.addTaskCompletionListener[Unit] { _ =>
+ val p = openArrayParser
+ if (p != null && !p.isClosed) p.close()
+ }
+ completionListenerRegistered = true
+ }
+ }
+ new Iterator[UTF8String] {
+ private var nextToken = parser.nextToken()
+
+ override def hasNext: Boolean = {
+ val more = nextToken != null && nextToken != JsonToken.END_ARRAY
+ if (!more) close()
+ more
+ }
+
+ override def next(): UTF8String = {
+ val element = serializeCurrentValue(parser)
+ nextToken = parser.nextToken()
+ element
+ }
+
+ // Close the parser and drop the evaluator's reference to it so the
listener does not retain
+ // it (and does not double-close) after this iterator is exhausted.
+ private def close(): Unit = {
+ if (!parser.isClosed) parser.close()
+ if (openArrayParser eq parser) openArrayParser = null
+ }
+ }
+ }
+
+ /**
+ * If `raw` is a JSON string literal (e.g. `"hi"`), returns its unquoted,
unescaped value;
+ * otherwise (numbers, booleans, objects, arrays) returns the raw JSON text
unchanged. Used to
+ * give a value column the string's content rather than its quoted JSON form.
+ *
+ * `raw` is a Jackson-serialized fragment with no leading whitespace, so
only a fragment whose
+ * first byte is `"` can be a string literal. Non-string values (the common
case) are returned
+ * without constructing a parser at all.
+ */
+ def unquotedString(raw: UTF8String): UTF8String = {
+ if (raw.numBytes() == 0 || raw.getByte(0) != '"') return raw
+ try {
+ Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, raw))
{ parser =>
+ if (parser.nextToken() == JsonToken.VALUE_STRING) {
+ UTF8String.fromString(parser.getText)
+ } else {
+ raw
+ }
+ }
+ } catch {
+ case _: JsonProcessingException => raw
+ }
+ }
+
+ /**
+ * Builds the prefix trie that lets [[navigateColumns]] resolve every column
path in one pass. The
+ * `paths` are indexed by result slot; a slot whose `include` is false (an
ordinality column,
+ * which has no JSON path) contributes nothing to the trie -- note this is
distinct from a root
+ * path `$`, which is an *empty but included* path that must resolve to the
whole item. Call once
+ * per `JSON_TABLE` invocation and reuse the result for every row.
+ */
+ def buildPathTrie(
+ paths: Array[Seq[PathInstruction]],
+ include: Array[Boolean]): JsonTablePathTrie = {
+ val root = new JsonTablePathTrie
+ var slot = 0
+ while (slot < paths.length) {
+ if (include(slot)) insertPath(root, paths(slot), slot)
+ slot += 1
+ }
+ root
+ }
+
+ private def insertPath(root: JsonTablePathTrie, path: Seq[PathInstruction],
slot: Int): Unit = {
+ var node = root
+ var rest = path
+ var valid = true
+ while (rest.nonEmpty && valid) {
+ rest match {
+ case Key :: Named(name) :: tail =>
+ node = node.named.getOrElseUpdate(name, new JsonTablePathTrie)
+ rest = tail
+ case Subscript :: Index(index) :: tail =>
+ node = node.indexed.getOrElseUpdate(index, new JsonTablePathTrie)
+ rest = tail
+ case _ =>
+ // Should not happen: JSON_TABLE column paths are validated to be
simple and
+ // wildcard-free. Drop the slot rather than mis-resolve it (it will
read as Missing).
+ valid = false
+ }
+ }
+ if (valid) node.terminals ::= slot
+ }
+
+ /**
+ * Resolves every column path (as built by [[buildPathTrie]]) within a
single row item in one
+ * traversal, returning the per-slot results. Slots for ordinality columns
(empty paths) are not
+ * in the trie and stay `Missing`; the caller fills them directly. Used to
extract value and
+ * EXISTS columns with correct missing-vs-null semantics.
+ */
+ def navigateColumns(item: UTF8String, trie: JsonTablePathTrie, numColumns:
Int)
+ : Array[JsonPathResult] = {
+ val out = Array.fill[JsonPathResult](numColumns)(JsonPathResult.Missing)
+ // No path columns (e.g. an ordinality-only table): skip parsing the item
entirely.
+ if (trie.isEmpty) return out
+ try {
+ Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, item))
{ parser =>
+ parser.nextToken()
+ navigateAll(parser, trie, out)
+ }
+ } catch {
+ // A malformed item leaves already-resolved slots in place and the rest
as Missing.
+ case _: JsonProcessingException =>
+ }
+ out
+ }
+}
+
/**
* The expression `GetJsonObject` will utilize it to support codegen.
*/
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 1621ec9ba9ac..349ac1300296 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -17,20 +17,22 @@
package org.apache.spark.sql.catalyst.expressions
+import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
CodeGenerator, CodegenFallback, ExprCode}
import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
import org.apache.spark.sql.catalyst.expressions.json.{GetJsonObjectEvaluator,
JsonExpressionUtils,
- JsonPathParser, JsonToStructsEvaluator, JsonTupleEvaluator,
MultiGetJsonObjectEvaluator,
- PathInstruction, SchemaOfJsonEvaluator, StructsToJsonEvaluator}
+ JsonPathParser, JsonPathResult, JsonTableEvaluator, JsonTablePathTrie,
JsonToStructsEvaluator,
+ JsonTupleEvaluator, MultiGetJsonObjectEvaluator, PathInstruction,
SchemaOfJsonEvaluator,
+ StructsToJsonEvaluator}
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.json._
import org.apache.spark.sql.catalyst.trees.TreePattern.{GET_JSON_OBJECT,
JSON_TO_STRUCT,
RUNTIME_REPLACEABLE, TreePattern}
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
import org.apache.spark.sql.types._
@@ -345,6 +347,292 @@ case class JsonTuple(children: Seq[Expression])
copy(children = newChildren)
}
+/**
+ * The kind of a single `JSON_TABLE` column.
+ */
+sealed trait JsonTableColumnKind
+object JsonTableColumnKind {
+ /** A `FOR ORDINALITY` column: a 1-based sequential row counter. */
+ case object Ordinality extends JsonTableColumnKind
+ /** A regular value column: extracts the value at `path` and casts it to
`dataType`. */
+ case object Value extends JsonTableColumnKind
+ /** An `EXISTS` column: true when `path` matches, cast to `dataType`. */
+ case object Exists extends JsonTableColumnKind
+}
+
+/**
+ * A single column definition of a `JSON_TABLE` invocation.
+ *
+ * @param name the output column name
+ * @param dataType the declared Spark type of the column (LongType for
ORDINALITY columns)
+ * @param path the SQL/JSON path relative to a row item; None for
ORDINALITY columns
+ * @param kind the column kind (ordinality / value / exists)
+ */
+case class JsonTableColumn(
+ name: String,
+ dataType: DataType,
+ path: Option[String],
+ kind: JsonTableColumnKind)
+
+/**
+ * Behavior when the JSON input is malformed.
+ */
+sealed trait JsonTableErrorMode
+object JsonTableErrorMode {
+ /** Produce no rows on malformed input (the SQL-standard default). */
+ case object NullOnError extends JsonTableErrorMode
+ /** Raise an error on malformed input. */
+ case object ErrorOnError extends JsonTableErrorMode
+}
+
+// scalastyle:off line.size.limit
+/**
+ * The SQL:2016 `JSON_TABLE` table-valued function. Shreds a JSON document
into a relational table:
+ * the `rowPath` selects a sequence of row items and each [[JsonTableColumn]]
projects a value out
+ * of each item. Implemented as a [[Generator]] so it plugs into the existing,
well-tested
+ * [[org.apache.spark.sql.catalyst.plans.logical.Generate]] operator; no new
execution operator is
+ * introduced.
+ *
+ * Only the flat (non-`NESTED PATH`) subset of the standard is supported.
Row-source and value
+ * extraction use the token-aware [[JsonTableEvaluator]], which (unlike
`get_json_object`)
+ * distinguishes a missing path from a JSON `null` value; type coercion reuses
[[Cast]].
+ *
+ * {{{
+ * SELECT t.* FROM json_table(
+ * '{"items":[{"id":1,"n":"a"},{"id":2,"n":"b"}]}',
+ * '$.items[*]'
+ * COLUMNS (seq FOR ORDINALITY, id INT PATH '$.id', name STRING PATH '$.n')
+ * ) AS t;
+ * }}}
+ */
+// scalastyle:on line.size.limit
+case class JsonTable(
+ child: Expression,
+ rowPath: String,
+ columns: Seq[JsonTableColumn],
+ errorMode: JsonTableErrorMode,
+ timeZoneId: Option[String] = None,
+ // Captured at plan-construction time so column casts do not change
behavior if the session's
+ // ANSI mode is flipped between building the plan and executing it
(matching `Cast`, which
+ // fixes its eval mode when the expression is constructed).
+ ansiEnabled: Boolean = SQLConf.get.ansiEnabled)
+ extends UnaryExpression
+ with Generator
+ with TimeZoneAwareExpression
+ with CodegenFallback
+ with ImplicitCastInputTypes
+ with QueryErrorsBase {
+
+ // Declared via ImplicitCastInputTypes so the analyzer coerces the JSON
input to STRING. In
+ // particular an untyped SQL NULL (NullType) is cast to STRING rather than
rejected, so
+ // `JSON_TABLE(NULL, ...)` reaches the runtime and applies the NULL ON ERROR
behavior.
+ override def inputTypes: Seq[AbstractDataType] =
+ Seq(StringTypeWithCollation(supportsTrimCollation = true))
+
+ // ORDINALITY columns always hold a non-null counter; value/EXISTS columns
may be null.
+ override def elementSchema: StructType =
+ StructType(columns.map { c =>
+ val nullable = c.kind != JsonTableColumnKind.Ordinality
+ StructField(c.name, c.dataType, nullable = nullable)
+ })
+
+ override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
+ copy(timeZoneId = Option(timeZoneId))
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ // First the standard input-type check (STRING for the JSON input, with
NULL coerced).
+ val inputCheck = super.checkInputDataTypes()
+ if (inputCheck.isFailure) {
+ inputCheck
+ } else {
+ // Validate the row path and every column path. A path is valid here iff
it parses and is
+ // free of wildcards -- except the row path may end in a single `[*]`,
which is stripped into
+ // `containerInstructions`, so the row path is checked on that
already-stripped list.
+ val rowPathValid = JsonPathParser.parse(rowPath).isDefined &&
+ !containerInstructions.contains(PathInstruction.Wildcard)
+ val invalid: Option[(String, String)] = if (!rowPathValid) {
+ Some(("row path", rowPath))
+ } else {
+ columns.iterator.collect { case c if c.path.isDefined => (c.name,
c.path.get) }
+ .collectFirst {
+ // Valid column path: parses and is wildcard-free, i.e.
hasWildcard == Some(false).
+ case (name, path) if
!JsonPathParser.hasWildcard(path).contains(false) =>
+ (s"column '$name'", path)
+ }
+ }
+ invalid match {
+ case Some((location, path)) =>
+ DataTypeMismatch(
+ errorSubClass = "INVALID_JSON_TABLE_PATH",
+ messageParameters = Map("location" -> location, "path" ->
toSQLValue(path)))
+ case None =>
+ // Every value/EXISTS column is produced by casting from a source
type (StringType for
+ // value columns, BooleanType for EXISTS columns) to the declared
column type. Reject a
+ // non-castable declared type (e.g. a value column declared
STRUCT/ARRAY/MAP) here rather
+ // than failing at runtime. Ordinality columns are always LongType
and need no check.
+ // The castability rules differ between ANSI and non-ANSI mode (e.g.
BOOLEAN -> TIMESTAMP
+ // is allowed by non-ANSI casts but not ANSI casts), so this must
use the same eval mode
+ // as the actual per-column `Cast` built in `columnCasts`.
+ def sourceType(c: JsonTableColumn): Option[DataType] = c.kind match {
+ case JsonTableColumnKind.Value => Some(StringType)
+ case JsonTableColumnKind.Exists => Some(BooleanType)
+ case JsonTableColumnKind.Ordinality => None
+ }
+ def castable(src: DataType, target: DataType): Boolean =
+ if (ansiEnabled) Cast.canAnsiCast(src, target) else
Cast.canCast(src, target)
+ columns.iterator.flatMap(c => sourceType(c).map((c, _)))
+ .collectFirst { case (c, src) if !castable(src, c.dataType) => (c,
src) } match {
+ case Some((c, srcType)) =>
+ DataTypeMismatch(
+ errorSubClass = "CAST_WITHOUT_SUGGESTION",
+ messageParameters = Map(
+ "srcType" -> toSQLType(srcType),
+ "targetType" -> toSQLType(c.dataType)))
+ case None =>
+ TypeCheckResult.TypeCheckSuccess
+ }
+ }
+ }
+ }
+
+ // The row path is `containerRowPath` plus an optional trailing `[*]`.
Splitting on the parsed
+ // instruction list (rather than the raw string) is whitespace-insensitive
and unambiguous.
+ // `checkInputDataTypes` guarantees the path parses and is wildcard-free at
this point.
+ @transient private lazy val (containerInstructions, explodeRoot)
+ : (Seq[PathInstruction], Boolean) = {
+ val parsed = JsonPathParser.parse(rowPath).getOrElse(Nil)
+ parsed match {
+ case init :+ PathInstruction.Subscript :+ PathInstruction.Wildcard =>
+ (init, true)
+ case other =>
+ (other, false)
+ }
+ }
+
+ @transient private lazy val rowEvaluator: JsonTableEvaluator =
+ JsonTableEvaluator(containerInstructions, explodeRoot)
+
+ // Parsed instruction list per column (empty for ordinality columns, which
have no path).
+ @transient private lazy val columnPaths: Array[Seq[PathInstruction]] =
+ columns.map(c =>
c.path.flatMap(JsonPathParser.parse).getOrElse(Nil)).toArray
+
+ // Column kinds snapshotted into an array, like `columnPaths` and
`columnCasts`: `columns` is a
+ // `List` (the parser builds it with `.map(...).toSeq`), so `columns(i)` is
O(i) and indexing it
+ // in the per-row projection loop would make `projectRow` O(n^2) in the
column count.
+ @transient private lazy val columnKinds: Array[JsonTableColumnKind] =
columns.map(_.kind).toArray
+
+ // Prefix trie over the column paths, built once so every row's value/EXISTS
columns are resolved
+ // in a single traversal of the item rather than one re-parse per column.
Ordinality columns have
+ // no path and are excluded (their empty path must not be confused with a
root path `$`, which is
+ // an included column reading the whole item).
+ @transient private lazy val columnTrie: JsonTablePathTrie = {
+ val include = columnKinds.map(_ != JsonTableColumnKind.Ordinality)
+ rowEvaluator.buildPathTrie(columnPaths, include)
+ }
+
+ // One reusable Cast per non-ordinality column, evaluated against a
single-slot mutable input
+ // row. Building the Cast once (over a BoundReference) avoids allocating an
expression tree per
+ // row/column on the hot path. The source type is BooleanType for EXISTS,
StringType otherwise.
+ @transient private lazy val columnCasts: Array[Expression] = {
+ val evalMode = EvalMode.fromBoolean(ansiEnabled)
+ columns.map { c =>
+ c.kind match {
+ case JsonTableColumnKind.Ordinality => null
+ case JsonTableColumnKind.Exists =>
+ Cast(BoundReference(0, BooleanType, nullable = false), c.dataType,
timeZoneId, evalMode)
+ case JsonTableColumnKind.Value =>
+ Cast(BoundReference(0, StringType, nullable = true), c.dataType,
timeZoneId, evalMode)
+ }
+ }.toArray
+ }
+
+ // Reusable single-slot input row for the per-column casts above.
+ @transient private lazy val castInput: GenericInternalRow = new
GenericInternalRow(1)
+
+ private def castColumn(i: Int, value: Any): Any = {
+ castInput.update(0, value)
+ columnCasts(i).eval(castInput)
+ }
+
+ private def projectRow(item: UTF8String, ordinal: Long): InternalRow = {
+ val numColumns = columnKinds.length
+ // Resolve every value/EXISTS column in a single traversal of the item;
ordinality slots are
+ // not in the trie and come back as Missing (filled below).
+ val resolved = rowEvaluator.navigateColumns(item, columnTrie, numColumns)
+ val values = new Array[Any](numColumns)
+ var i = 0
+ while (i < numColumns) {
+ values(i) = columnKinds(i) match {
+ case JsonTableColumnKind.Ordinality =>
+ ordinal
+ case JsonTableColumnKind.Exists =>
+ // Present (including an explicit JSON null) counts as existing;
only Missing is false.
+ val exists = resolved(i) != JsonPathResult.Missing
+ castColumn(i, exists)
+ case JsonTableColumnKind.Value =>
+ resolved(i) match {
+ // `raw` is a re-parseable JSON fragment; unquote a scalar string
so the column gets
+ // its content (e.g. `"hi"` -> `hi`), then cast to the declared
type.
+ case JsonPathResult.Found(raw) => castColumn(i,
rowEvaluator.unquotedString(raw))
+ // A missing path and an explicit JSON null both yield SQL NULL
for a value column.
+ case _ => null
+ }
+ }
+ i += 1
+ }
+ new GenericInternalRow(values)
+ }
+
+ override def eval(input: InternalRow): IterableOnce[InternalRow] = {
+ val json = child.eval(input).asInstanceOf[UTF8String]
+ rowEvaluator.evaluate(json) match {
+ case Some(items) =>
+ // A manual Long counter for FOR ORDINALITY: `zipWithIndex` is
Int-based and would wrap
+ // past Int.MaxValue for a very large array, whereas ordinality is a
BIGINT.
+ var ordinal = 0L
+ items.map { item =>
+ ordinal += 1L
+ projectRow(item, ordinal)
+ }
+ case None =>
+ // `errorMode` governs the row-source JSON only (null / malformed
input, or `[*]` over a
+ // non-array). Per-column value extraction follows normal `Cast`
semantics: a bad cast is
+ // NULL in non-ANSI mode and raises in ANSI mode, independent of ON
ERROR.
+ errorMode match {
+ case JsonTableErrorMode.NullOnError => Iterator.empty
+ case JsonTableErrorMode.ErrorOnError =>
+ throw
QueryExecutionErrors.malformedRecordsDetectedInRecordParsingError(
+ if (json == null) "null" else json.toString,
+ SparkException.internalError("JSON_TABLE encountered malformed
JSON input."))
+ }
+ }
+ }
+
+ override def prettyName: String = "json_table"
+
+ // The default `Expression.sql` renders only children, i.e.
`json_table(<json_expr>)`, dropping
+ // the row path, columns, and ON ERROR mode. Render the full
`JSON_TABLE(...)` syntax so
+ // analysis/type-check diagnostics (e.g. INVALID_JSON_TABLE_PATH) point at
the whole invocation.
+ override def sql: String = {
+ val columnsSQL = columns.map { c =>
+ val pathSQL = c.path.map(p => s" PATH '$p'").getOrElse("")
+ c.kind match {
+ case JsonTableColumnKind.Ordinality => s"${c.name} FOR ORDINALITY"
+ case JsonTableColumnKind.Exists => s"${c.name} ${c.dataType.sql}
EXISTS$pathSQL"
+ case JsonTableColumnKind.Value => s"${c.name}
${c.dataType.sql}$pathSQL"
+ }
+ }.mkString(", ")
+ val errorSQL = errorMode match {
+ case JsonTableErrorMode.NullOnError => "NULL ON ERROR"
+ case JsonTableErrorMode.ErrorOnError => "ERROR ON ERROR"
+ }
+ s"JSON_TABLE(${child.sql}, '$rowPath' COLUMNS ($columnsSQL) $errorSQL)"
+ }
+
+ override protected def withNewChildInternal(newChild: Expression): JsonTable
=
+ copy(child = newChild)
+}
+
/**
* Converts an json input string to a [[StructType]], [[ArrayType]] or
[[MapType]]
* with the specified schema.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index 611bd86ea9bd..534f87c1e7eb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2069,6 +2069,7 @@ class AstBuilder extends DataTypeAstBuilder
case _: AliasedQueryContext =>
case _: TableValuedFunctionContext =>
case _: UnnestTableContext =>
+ case _: JsonTableRelationContext =>
case other =>
throw QueryParsingErrors.invalidLateralJoinRelationError(other)
}
@@ -2554,6 +2555,7 @@ class AstBuilder extends DataTypeAstBuilder
case _: AliasedQueryContext =>
case _: TableValuedFunctionContext =>
case _: UnnestTableContext =>
+ case _: JsonTableRelationContext =>
case other =>
throw QueryParsingErrors.invalidLateralJoinRelationError(other)
}
@@ -3241,6 +3243,95 @@ 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. Whether two
names that differ only
+ // in case collide follows the configured resolver, so `a` and `A` stay
distinct under
+ // `spark.sql.caseSensitive`.
+ val normalize: String => String =
+ if (conf.caseSensitiveAnalysis) identity else _.toLowerCase(Locale.ROOT)
+ val duplicate = columns.groupBy(c => normalize(c.name)).collectFirst {
+ case (_, cols) if cols.length > 1 => cols.head.name
+ }
+ duplicate.foreach { name =>
+ throw QueryParsingErrors.duplicateJsonTableColumnError(name, jt)
+ }
+
+ val errorMode = if (jt.jsonTableOnErrorClause != null &&
jt.jsonTableOnErrorClause.ERROR != null
+ && jt.jsonTableOnErrorClause.NULL == null) {
+ JsonTableErrorMode.ErrorOnError
+ } else {
+ JsonTableErrorMode.NullOnError
+ }
+
+ val generator = JsonTable(jsonExpr, rowPath, columns, errorMode)
+ val generate = Generate(
+ generator,
+ unrequiredChildIndex = Nil,
+ outer = false,
+ qualifier = None,
+ generatorOutput = columns.map(c => UnresolvedAttribute.quoted(c.name)),
+ child = OneRowRelation())
+ mayApplyAliasPlan(jt.tableAlias, generate)
+ }
+
+ /**
+ * The implicit JSON path for a column with no explicit PATH: the column
name as a single JSON
+ * object key. Bracket syntax (`$['name']`) is used rather than `$.name` so
a column name that
+ * contains a dot (e.g. `a.b`) reads the literal key `"a.b"` instead of the
nested path `a.b`. A
+ * name containing a single quote cannot be represented and yields an
unparseable path, which
+ * `JsonTable.checkInputDataTypes` rejects (such a column must use an
explicit PATH).
+ */
+ private def implicitJsonTablePath(name: String): String = s"$$['$name']"
+
+ /**
+ * Build a single [[JsonTableColumn]] from a `jsonTableColumn` grammar
context. A value column
+ * with no explicit PATH gets an implicit path derived from its name (see
+ * [[implicitJsonTablePath]]), matching the SQL standard / Oracle behavior.
+ */
+ private def buildJsonTableColumn(ctx: JsonTableColumnContext):
JsonTableColumn = withOrigin(ctx) {
+ ctx match {
+ case ord: JsonTableOrdinalityColumnContext =>
+ JsonTableColumn(
+ name = getIdentifierText(ord.colName),
+ dataType = LongType,
+ path = None,
+ kind = JsonTableColumnKind.Ordinality)
+ case ex: JsonTableExistsColumnContext =>
+ val name = getIdentifierText(ex.colName)
+ val path = Option(ex.path).map(p => string(visitStringLit(p)))
+ .getOrElse(implicitJsonTablePath(name))
+ JsonTableColumn(
+ name = name,
+ // A column value is produced by a `Cast` to the declared type, so
normalize CHAR/VARCHAR
+ // to STRING exactly as `visitCast` does; a raw CHAR/VARCHAR target
has no encoder.
+ dataType = CharVarcharUtils.replaceCharVarcharWithStringForCast(
+ typedVisit[DataType](ex.dataType)),
+ path = Some(path),
+ kind = JsonTableColumnKind.Exists)
+ case v: JsonTableValueColumnContext =>
+ val name = getIdentifierText(v.colName)
+ val path = Option(v.path).map(p => string(visitStringLit(p)))
+ .getOrElse(implicitJsonTablePath(name))
+ JsonTableColumn(
+ name = name,
+ dataType = CharVarcharUtils.replaceCharVarcharWithStringForCast(
+ typedVisit[DataType](v.dataType)),
+ path = Some(path),
+ kind = JsonTableColumnKind.Value)
+ }
+ }
+
/**
* Extract the source name from an identifiedByClause context.
*/
diff --git
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
index 2df258a2dc5c..b305541893bb 100644
---
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
+++
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
@@ -210,7 +210,7 @@ class SparkConnectDatabaseMetaDataSuite extends
ConnectFunSuite with RemoteSpark
val metadata = conn.getMetaData
// scalastyle:off line.size.limit
// CURRENT_PATH and SYSTEM are excluded: getSQLKeywords drops SQL:2003
reserved words (see companion).
- assert(metadata.getSQLKeywords ===
"ADD,AFTER,AGGREGATE,ALIGN,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPLY,APPROX,ARCHIVE,ASC,ASOF,AUTO,BERNOULLI,BIN,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF
[...]
+ assert(metadata.getSQLKeywords ===
"ADD,AFTER,AGGREGATE,ALIGN,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPLY,APPROX,ARCHIVE,ASC,ASOF,AUTO,BERNOULLI,BIN,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF
[...]
// scalastyle:on line.size.limit
}
}
diff --git
a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
index 943482bece87..fe26fef775ca 100644
--- a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
@@ -132,6 +132,7 @@ ELSE true
ELSEIF false
END true
ENFORCED false
+ERROR false
ESCAPE true
ESCAPED false
EVOLUTION false
@@ -210,6 +211,7 @@ ITEMS false
ITERATE false
JOIN true
JSON false
+JSON_TABLE false
KEY false
KEYS false
LANGUAGE false
diff --git a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
index c180dd97c573..d23be26a60dd 100644
--- a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
@@ -132,6 +132,7 @@ ELSE false
ELSEIF false
END false
ENFORCED false
+ERROR false
ESCAPE false
ESCAPED false
EVOLUTION false
@@ -210,6 +211,7 @@ ITEMS false
ITERATE false
JOIN false
JSON false
+JSON_TABLE false
KEY false
KEYS false
LANGUAGE false
diff --git
a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
index c180dd97c573..d23be26a60dd 100644
--- a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
@@ -132,6 +132,7 @@ ELSE false
ELSEIF false
END false
ENFORCED false
+ERROR false
ESCAPE false
ESCAPED false
EVOLUTION false
@@ -210,6 +211,7 @@ ITEMS false
ITERATE false
JOIN false
JSON false
+JSON_TABLE false
KEY false
KEYS false
LANGUAGE false
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JsonTableSuite.scala
new file mode 100644
index 000000000000..3772cffc39d1
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonTableSuite.scala
@@ -0,0 +1,800 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType,
TimestampType}
+
+/**
+ * End-to-end tests for the SQL:2016 `JSON_TABLE` table-valued function (flat,
non-nested subset).
+ */
+class JsonTableSuite extends QueryTest with SharedSparkSession {
+
+ test("expand a JSON array into rows with typed columns and ordinality") {
+ val json =
"""{"items":[{"id":1,"n":"a"},{"id":2,"n":"b"},{"id":3,"n":"c"}]}"""
+ val df = sql(
+ s"""
+ |SELECT t.* FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (
+ | seq FOR ORDINALITY,
+ | id INT PATH '$$.id',
+ | name STRING PATH '$$.n'
+ | )
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1L, 1, "a"), Row(2L, 2, "b"), Row(3L, 3, "c")))
+ // Ordinality is a BIGINT, typed columns take their declared types.
+ assert(df.schema.map(_.dataType) === Seq(LongType, IntegerType,
StringType))
+ }
+
+ test("implicit column path derived from column name") {
+ val json = """{"rows":[{"id":10,"name":"x"},{"id":20,"name":"y"}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.rows[*]'
+ | COLUMNS (id INT, name STRING)
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(10, "x"), Row(20, "y")))
+ }
+
+ test("row path matching a single object yields one row") {
+ val json = """{"a":1,"b":"hello"}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$'
+ | COLUMNS (a INT PATH '$$.a', b STRING PATH '$$.b')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1, "hello")))
+ }
+
+ test("missing column path yields null") {
+ val json = """{"items":[{"id":1},{"id":2,"n":"b"}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (id INT PATH '$$.id', name STRING PATH '$$.n')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1, null), Row(2, "b")))
+ }
+
+ test("EXISTS column reports presence as boolean") {
+ val json = """{"items":[{"id":1,"opt":9},{"id":2}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (id INT PATH '$$.id', hasOpt BOOLEAN EXISTS PATH '$$.opt')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1, true), Row(2, false)))
+ }
+
+ test("empty array yields no rows") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[]}',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq.empty)
+ }
+
+ test("row path matching nothing yields no rows") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]}',
+ | '$.absent[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq.empty)
+ }
+
+ test("NULL ON ERROR (default) yields no rows for malformed JSON") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{ this is not valid json',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq.empty)
+
+ // Explicit NULL ON ERROR behaves the same.
+ val df2 = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{ this is not valid json',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ | NULL ON ERROR
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df2, Seq.empty)
+ }
+
+ test("ERROR ON ERROR raises for malformed JSON") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{ this is not valid json',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ | ERROR ON ERROR
+ |) AS t
+ """.stripMargin)
+ intercept[SparkException] {
+ df.collect()
+ }
+ }
+
+ test("null JSON input yields no rows") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | CAST(NULL AS STRING),
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq.empty)
+ }
+
+ test("untyped NULL input is coerced and yields no rows (NULL ON ERROR)") {
+ // An untyped SQL NULL (NullType) must be coerced to STRING and apply NULL
ON ERROR, not be
+ // rejected during analysis.
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | NULL,
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq.empty)
+ }
+
+ test("join JSON_TABLE output against a base table") {
+ import testImplicits._
+ withTempView("docs") {
+ Seq(
+ (1, """{"tags":[{"k":"a"},{"k":"b"}]}"""),
+ (2, """{"tags":[{"k":"c"}]}""")
+ ).toDF("id", "doc").createOrReplaceTempView("docs")
+
+ val df = sql(
+ """
+ |SELECT d.id, t.k
+ |FROM docs d,
+ |LATERAL json_table(d.doc, '$.tags[*]' COLUMNS (k STRING PATH
'$.k')) AS t
+ |ORDER BY d.id, t.k
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1, "a"), Row(1, "b"), Row(2, "c")))
+ }
+ }
+
+ test("nested field extraction within a row item") {
+ val json = """{"items":[{"meta":{"score":7}},{"meta":{"score":8}}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (score INT PATH '$$.meta.score')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(7), Row(8)))
+ }
+
+ test("column path that is a prefix of another is resolved correctly") {
+ // `$.meta` both terminates a column and is a prefix of `$.meta.score` and
`$.meta.name`, so
+ // all three (plus an EXISTS on the prefix) must resolve from the same
item.
+ val json =
"""{"items":[{"meta":{"score":7,"name":"a"}},{"meta":{"score":8,"name":"b"}}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (
+ | meta STRING PATH '$$.meta',
+ | has_meta BOOLEAN EXISTS PATH '$$.meta',
+ | score INT PATH '$$.meta.score',
+ | name STRING PATH '$$.meta.name')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(
+ Row("""{"score":7,"name":"a"}""", true, 7, "a"),
+ Row("""{"score":8,"name":"b"}""", true, 8, "b")))
+ }
+
+ test("ordinality-only table produces one numbered row per array element") {
+ // No path columns: every element still yields a row, numbered by
ordinality, even when the
+ // element itself is a scalar (the item is never inspected for a value).
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[10,20,30]}',
+ | '$.items[*]'
+ | COLUMNS (seq FOR ORDINALITY)
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1L), Row(2L), Row(3L)))
+ }
+
+ test("column alias list from the table alias") {
+ val json = """{"items":[{"id":1,"n":"a"}]}"""
+ val df = sql(
+ s"""
+ |SELECT renamed_id, renamed_name FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (id INT PATH '$$.id', name STRING PATH '$$.n')
+ |) AS t(renamed_id, renamed_name)
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1, "a")))
+ }
+
+ test("duplicate column names are rejected at parse time") {
+ val e = intercept[ParseException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]}',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id', id STRING PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ }
+ assert(e.getCondition == "INVALID_SQL_SYNTAX.DUPLICATE_JSON_TABLE_COLUMN")
+ assert(e.getMessageParameters.get("columnName") == "`id`")
+ }
+
+ test("duplicate column name detection follows spark.sql.caseSensitive") {
+ // Names differing only in case: distinct under a case-sensitive resolver,
colliding otherwise.
+ val mixedCase =
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1,"ID":2}]}',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id', ID INT PATH '$.ID')
+ |) AS t
+ """.stripMargin
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ checkAnswer(sql(mixedCase), Seq(Row(1, 2)))
+ }
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val e = intercept[ParseException](sql(mixedCase))
+ assert(e.getCondition ==
"INVALID_SQL_SYNTAX.DUPLICATE_JSON_TABLE_COLUMN")
+ assert(e.getMessageParameters.get("columnName") == "`id`")
+ }
+ // An exact-case duplicate is rejected in both modes.
+ val exactDuplicate =
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]}',
+ | '$.items[*]'
+ | COLUMNS (id INT PATH '$.id', id INT PATH '$.id')
+ |) AS t
+ """.stripMargin
+ Seq("true", "false").foreach { caseSensitive =>
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) {
+ val e = intercept[ParseException](sql(exactDuplicate))
+ assert(e.getCondition ==
"INVALID_SQL_SYNTAX.DUPLICATE_JSON_TABLE_COLUMN")
+ }
+ }
+ }
+
+ test("value cast honors ANSI mode") {
+ val json = """{"items":[{"v":"not_a_number"}]}"""
+ val query =
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (v INT PATH '$$.v')
+ |) AS t
+ """.stripMargin
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ checkAnswer(sql(query), Seq(Row(null)))
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ // ANSI cast failures surface as a SparkThrowable (e.g.
SparkNumberFormatException).
+ intercept[SparkThrowable] {
+ sql(query).collect()
+ }
+ }
+ }
+
+ test("EXISTS distinguishes a present JSON null from a missing key") {
+ // A key present with a JSON null value EXISTS (true); a truly absent key
does not (false).
+ val json = """{"items":[{"a":null},{"b":1}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (hasA BOOLEAN EXISTS PATH '$$.a')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(true), Row(false)))
+ }
+
+ test("value column returns SQL NULL (not the string 'null') for a JSON
null") {
+ // JSON null must become SQL NULL, while a JSON string "null" must remain
the string.
+ val json = """{"items":[{"v":null},{"v":"null"},{"v":"x"}]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (v STRING PATH '$$.v')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(null), Row("null"), Row("x")))
+ }
+
+ test("value column returns SQL NULL for a JSON null reached via an array
index") {
+ val json = """{"arr":[null, "kept"]}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$'
+ | COLUMNS (first STRING PATH '$$.arr[0]', second STRING PATH
'$$.arr[1]')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(null, "kept")))
+ }
+
+ test("mid-path wildcard in a column or row path is rejected") {
+ // A wildcard anywhere except a single trailing '[*]' on the row path is
unsupported.
+ val e1 = intercept[AnalysisException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"x":1}]}',
+ | '$.items[*].x'
+ | COLUMNS (x INT PATH '$.x')
+ |) AS t
+ """.stripMargin).collect()
+ }
+ assert(e1.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_TABLE_PATH")
+ assert(e1.getMessageParameters.get("location") == "row path")
+ assert(e1.getMessageParameters.get("path") == "'$.items[*].x'")
+ // The rendered expression carries the full JSON_TABLE syntax (row path +
columns + ON ERROR),
+ // not just the JSON input, so the diagnostic is actionable.
+ assert(e1.getMessage.contains(
+ "JSON_TABLE({\"items\":[{\"x\":1}]}, '$.items[*].x' " +
+ "COLUMNS (x INT PATH '$.x') NULL ON ERROR)"))
+
+ val e2 = intercept[AnalysisException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"vals":[1,2]}]}',
+ | '$.items[*]'
+ | COLUMNS (v INT PATH '$.vals[*]')
+ |) AS t
+ """.stripMargin).collect()
+ }
+ assert(e2.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_TABLE_PATH")
+ assert(e2.getMessageParameters.get("location") == "column 'v'")
+ assert(e2.getMessageParameters.get("path") == "'$.vals[*]'")
+ }
+
+ test("[*] over a non-array fires ERROR ON ERROR and is empty under NULL ON
ERROR") {
+ val json = """{"items":{"a":1}}"""
+ // NULL ON ERROR (default): a non-array under [*] yields no rows.
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json', '$$.items[*]' COLUMNS (a INT PATH '$$.a')
+ |) AS t
+ """.stripMargin),
+ Seq.empty)
+ // ERROR ON ERROR: the same input raises.
+ intercept[SparkException] {
+ sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json', '$$.items[*]' COLUMNS (a INT PATH '$$.a') ERROR ON ERROR
+ |) AS t
+ """.stripMargin).collect()
+ }
+ }
+
+ test("non-explode row path resolving to a string value") {
+ // The matched row item is a JSON string; a value column reading '$' must
get its content.
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"name":"hello world"}',
+ | '$.name'
+ | COLUMNS (c STRING PATH '$', present BOOLEAN EXISTS PATH '$')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row("hello world", true)))
+ }
+
+ test("row path resolving to a top-level string") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table('"just a string"', '$' COLUMNS (c STRING
PATH '$')) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row("just a string")))
+ }
+
+ test("array of scalars as the row source") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"nums":[1,2,3]}',
+ | '$.nums[*]'
+ | COLUMNS (seq FOR ORDINALITY, v INT PATH '$')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1L, 1), Row(2L, 2), Row(3L, 3)))
+ }
+
+ test("high-precision fractional value is preserved exactly, not rounded to a
double") {
+ // The matched fragment is reserialized before JSON_TABLE casts it. The
number carries more
+ // significant digits than a double can hold, so serializing via a lossy
float copy would round
+ // it before the DECIMAL/STRING cast sees it. Exact structure copying
keeps the digits verbatim.
+ val json = """{"v":123456789.123456789123456789}"""
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json',
+ | '$$'
+ | COLUMNS (d DECIMAL(38, 18) PATH '$$.v', s STRING PATH '$$.v')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df,
+ Seq(Row(BigDecimal("123456789.123456789123456789"),
"123456789.123456789123456789")))
+ }
+
+ test("array of strings as the row source keeps string content") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"tags":["a","b c"]}',
+ | '$.tags[*]'
+ | COLUMNS (v STRING PATH '$')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row("a"), Row("b c")))
+ }
+
+ test("duplicate keys within an item resolve to the first value") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"a":1,"a":2}]}',
+ | '$.items[*]'
+ | COLUMNS (a INT PATH '$.a')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1)))
+ }
+
+ test("structure-valued column serialized to STRING") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"o":{"x":1}}]}',
+ | '$.items[*]'
+ | COLUMNS (o STRING PATH '$.o')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row("""{"x":1}""")))
+ }
+
+ test("unsupported declared column type is rejected at analysis") {
+ // A value column cannot be declared as a complex type that STRING cannot
be cast to.
+ val e = intercept[AnalysisException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"v":1}]}',
+ | '$.items[*]'
+ | COLUMNS (v STRUCT<a: INT> PATH '$.v')
+ |) AS t
+ """.stripMargin).collect()
+ }
+ assert(e.getCondition == "DATATYPE_MISMATCH.CAST_WITHOUT_SUGGESTION")
+ }
+
+ test("oversized numeric path index is rejected as an invalid path") {
+ val e = intercept[AnalysisException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"a":[1]}',
+ | '$'
+ | COLUMNS (v INT PATH '$.a[999999999999999999999999]')
+ |) AS t
+ """.stripMargin).collect()
+ }
+ assert(e.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_TABLE_PATH")
+ }
+
+ test("ERROR ON ERROR rejects trailing garbage and empty input") {
+ // A valid JSON prefix followed by garbage must raise, not silently
produce rows.
+ intercept[SparkException] {
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]} trailing garbage',
+ | '$.items[*]' COLUMNS (id INT PATH '$.id') ERROR ON ERROR
+ |) AS t
+ """.stripMargin).collect()
+ }
+ // Empty input is malformed under ERROR ON ERROR.
+ intercept[SparkException] {
+ sql(
+ """
+ |SELECT * FROM json_table('', '$.items[*]' COLUMNS (id INT PATH
'$.id') ERROR ON ERROR)
+ |AS t
+ """.stripMargin).collect()
+ }
+ // Under the default NULL ON ERROR the same inputs simply yield no rows.
+ checkAnswer(
+ sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]} trailing garbage',
+ | '$.items[*]' COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin),
+ Seq.empty)
+ }
+
+ test("large array row source is expanded correctly") {
+ // A big array is fully and correctly expanded. Row emission is streamed
element by element
+ // from the source parser (the whole expanded payload is not materialized
at once); note the
+ // input is still scanned once up front to validate it is a single
well-formed JSON document.
+ val n = 5000
+ val arr = (1 to n).map(i => s"""{"id":$i}""").mkString(",")
+ val json = s"""{"items":[$arr]}"""
+ val df = sql(
+ s"""
+ |SELECT count(*) AS c, sum(id) AS s FROM json_table(
+ | '$json',
+ | '$$.items[*]'
+ | COLUMNS (id INT PATH '$$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(n.toLong, (n.toLong * (n + 1)) / 2)))
+
+ // A bare LIMIT (no ORDER BY) stops pulling generator rows early, so the
streaming iterator is
+ // abandoned before exhaustion -- this exercises the
task-completion-listener parser cleanup.
+ // Elements are in document order, so the first three ids are 1, 2, 3.
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT id FROM json_table('$json', '$$.items[*]' COLUMNS (id INT
PATH '$$.id')) AS t
+ |LIMIT 3
+ """.stripMargin),
+ Seq(Row(1), Row(2), Row(3)))
+ }
+
+ test("deep container path to an array row source") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"a":{"b":[{"id":1},{"id":2}]}}',
+ | '$.a.b[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ }
+
+ test("indexed container path then wildcard") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"a":[{"vals":[{"id":10},{"id":20}]}]}',
+ | '$.a[0].vals[*]'
+ | COLUMNS (id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(10), Row(20)))
+ }
+
+ test("non-explode row path resolving to JSON null yields one row of nulls") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"x":null}',
+ | '$.x'
+ | COLUMNS (v STRING PATH '$', present BOOLEAN EXISTS PATH '$')
+ |) AS t
+ """.stripMargin)
+ // The row source is a JSON null: one row; a value column is SQL NULL,
EXISTS is true.
+ checkAnswer(df, Seq(Row(null, true)))
+ }
+
+ test("[*] over a JSON null container yields no rows / errors per ON ERROR") {
+ val json = """{"items":null}"""
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT * FROM json_table('$json', '$$.items[*]' COLUMNS (id INT
PATH '$$.id')) AS t
+ """.stripMargin),
+ Seq.empty)
+ intercept[SparkException] {
+ sql(
+ s"""
+ |SELECT * FROM json_table(
+ | '$json', '$$.items[*]' COLUMNS (id INT PATH '$$.id') ERROR ON
ERROR
+ |) AS t
+ """.stripMargin).collect()
+ }
+ }
+
+ test("implicit path for a column name containing a dot reads the literal
key") {
+ // `a.b` with no PATH must read the JSON key "a.b", not the nested path a
-> b.
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"a.b":1, "a":{"b":2}}]}',
+ | '$.items[*]'
+ | COLUMNS (`a.b` INT)
+ |) AS t
+ """.stripMargin)
+ checkAnswer(df, Seq(Row(1)))
+ }
+
+ test("column cast eval mode is captured at plan construction, not
execution") {
+ // Build the DataFrame with ANSI off (bad cast -> NULL); enabling ANSI
afterwards must not
+ // change the already-planned generator's behavior.
+ val query =
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"v":"not_a_number"}]}',
+ | '$.items[*]'
+ | COLUMNS (v INT PATH '$.v')
+ |) AS t
+ """.stripMargin
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ val df = sql(query)
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ checkAnswer(df, Seq(Row(null)))
+ }
+ }
+ }
+
+ test("FOR ORDINALITY column is non-nullable in the output schema") {
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"id":1}]}',
+ | '$.items[*]'
+ | COLUMNS (seq FOR ORDINALITY, id INT PATH '$.id')
+ |) AS t
+ """.stripMargin)
+ assert(!df.schema("seq").nullable)
+ assert(df.schema("id").nullable)
+ }
+
+ test("castability check uses the session ANSI mode") {
+ // BOOLEAN -> TIMESTAMP is castable in non-ANSI mode but not in ANSI mode,
so an
+ // `EXISTS ... TIMESTAMP` column must be accepted under non-ANSI and
rejected under ANSI,
+ // matching the eval mode of the actual per-column Cast.
+ val query =
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"a":1}]}',
+ | '$.items[*]'
+ | COLUMNS (hasA TIMESTAMP EXISTS PATH '$.a')
+ |) AS t
+ """.stripMargin
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ // Accepted at analysis; BOOLEAN true casts to a timestamp value.
+ assert(sql(query).schema("hasA").dataType == TimestampType)
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ val e = intercept[AnalysisException](sql(query).collect())
+ assert(e.getCondition == "DATATYPE_MISMATCH.CAST_WITHOUT_SUGGESTION")
+ }
+ }
+
+ test("CHAR/VARCHAR column types are normalized to STRING") {
+ // A raw CHAR/VARCHAR target has no runtime encoder; like a normal CAST,
the declared type is
+ // normalized to STRING so the column is produced and queried without
error.
+ val df = sql(
+ """
+ |SELECT * FROM json_table(
+ | '{"items":[{"c":"hi","v":"world"}]}',
+ | '$.items[*]'
+ | COLUMNS (c CHAR(5) PATH '$.c', v VARCHAR(10) PATH '$.v')
+ |) AS t
+ """.stripMargin)
+ assert(df.schema("c").dataType == StringType)
+ assert(df.schema("v").dataType == StringType)
+ checkAnswer(df, Seq(Row("hi", "world")))
+ }
+
+ test("wide projection interleaving all column kinds keeps every column in
its own slot") {
+ // The per-row projection reads column kinds, paths, casts and trie slots
from parallel arrays,
+ // so a wide projection that interleaves the three kinds (and leaves some
paths unmatched)
+ // pins each column to its own slot.
+ val n = 60
+ val fields = (1 to n).map(i => s""""f$i":$i""").mkString(",")
+ val columns = (1 to n).map { i =>
+ i % 3 match {
+ case 0 => s"ord$i FOR ORDINALITY"
+ case 1 => s"val$i INT PATH '$$.f$i'"
+ // Some EXISTS columns point at a key the item does not have, so both
outcomes are covered.
+ case _ if i % 4 == 0 => s"ex$i BOOLEAN EXISTS PATH '$$.f$i'"
+ case _ => s"ex$i BOOLEAN EXISTS PATH '$$.missing$i'"
+ }
+ }.mkString(", ")
+ val df = sql(
+ s"""
+ |SELECT * FROM json_table('{"items":[{$fields}]}', '$$.items[*]'
COLUMNS ($columns)) AS t
+ """.stripMargin)
+ val expected = (1 to n).map { i =>
+ i % 3 match {
+ case 0 => 1L // single row, so ordinality is 1
+ case 1 => i
+ case _ => i % 4 == 0
+ }
+ }
+ checkAnswer(df, Seq(Row.fromSeq(expected)))
+ }
+
+ test("array row source over many input rows streams without leaking
parsers") {
+ // Exercises the per-task (not per-row) parser cleanup: many input rows,
each with a `[*]`
+ // array row source, must all shred correctly.
+ withTempView("docs") {
+ spark.range(0, 200)
+ .selectExpr("concat('{\"items\":[{\"v\":', id, '},{\"v\":', id, '}]}')
AS j")
+ .createOrReplaceTempView("docs")
+ val df = sql(
+ """
+ |SELECT t.v
+ |FROM docs d,
+ |LATERAL json_table(d.j, '$.items[*]' COLUMNS (v INT PATH '$.v')) AS
t
+ """.stripMargin)
+ // Two rows per input document.
+ assert(df.count() == 400)
+ }
+ }
+}
diff --git
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
index 4dd2948988d6..c285023f82b0 100644
---
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
+++
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
@@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends
SharedThriftServer {
val sessionHandle = client.openSession(user, "")
val infoValue = client.getInfo(sessionHandle,
GetInfoType.CLI_ODBC_KEYWORDS)
// scalastyle:off line.size.limit
- assert(infoValue.getStringValue ==
"ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,ASOF,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,C
[...]
+ assert(infoValue.getStringValue ==
"ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,ASOF,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,C
[...]
// scalastyle:on line.size.limit
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
