diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 6df35deef807..16b3debba46a 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1963,6 +1963,16 @@
           "<functionName> has an invalid or unsupported JSON path <path>. Only simple, wildcard-free paths are supported."
         ]
       },
+      "INVALID_JSON_QUERY_RETURNING_TYPE" : {
+        "message" : [
+          "<functionName> cannot return a value of type <returningType>. The RETURNING type must be a string type."
+        ]
+      },
+      "INVALID_JSON_QUERY_WRAPPER_AND_QUOTES" : {
+        "message" : [
+          "<functionName> cannot combine an OMIT QUOTES clause with a WITH ARRAY WRAPPER clause. OMIT QUOTES applies only to an unwrapped result."
+        ]
+      },
       "INVALID_JSON_SCALAR_RETURNING_TYPE" : {
         "message" : [
           "<functionName> cannot return a value of type <returningType>. The RETURNING type must be a scalar (string, numeric, boolean, or datetime) type."
@@ -5589,6 +5599,24 @@
     ],
     "sqlState" : "42K0E"
   },
+  "JSON_QUERY_ON_ERROR" : {
+    "message" : [
+      "<functionName> could not extract a value at path <path>."
+    ],
+    "subClass" : {
+      "EMPTY" : {
+        "message" : [
+          "The path matched no value. This error was requested by the ERROR ON EMPTY clause."
+        ]
+      },
+      "ERROR" : {
+        "message" : [
+          "The input is not valid JSON. This error was requested by the ERROR ON ERROR clause."
+        ]
+      }
+    },
+    "sqlState" : "2203G"
+  },
   "JSON_VALUE_ON_ERROR" : {
     "message" : [
       "<functionName> could not extract a scalar value at path <path>."
diff --git a/docs/sql-ref-ansi-compliance.md b/docs/sql-ref-ansi-compliance.md
index 77c4d96da1ea..3a40337f7e9f 100644
--- a/docs/sql-ref-ansi-compliance.md
+++ b/docs/sql-ref-ansi-compliance.md
@@ -484,6 +484,7 @@ Below is a list of all the keywords in Spark SQL.
 |COMPUTE|non-reserved|non-reserved|non-reserved|
 |CONCATENATE|non-reserved|non-reserved|non-reserved|
 |CONDITION|non-reserved|non-reserved|non-reserved|
+|CONDITIONAL|non-reserved|non-reserved|non-reserved|
 |CONSTRAINT|reserved|non-reserved|reserved|
 |CONTAINS|non-reserved|non-reserved|non-reserved|
 |CONTINUE|non-reserved|non-reserved|non-reserved|
@@ -618,8 +619,10 @@ 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_QUERY|non-reserved|non-reserved|reserved|
 |JSON_TABLE|non-reserved|non-reserved|reserved|
 |JSON_VALUE|non-reserved|non-reserved|reserved|
+|KEEP|non-reserved|non-reserved|non-reserved|
 |KEY|non-reserved|non-reserved|non-reserved|
 |KEYS|non-reserved|non-reserved|non-reserved|
 |LANGUAGE|non-reserved|non-reserved|reserved|
@@ -679,8 +682,10 @@ Below is a list of all the keywords in Spark SQL.
 |NULL|reserved|non-reserved|reserved|
 |NULLS|non-reserved|non-reserved|non-reserved|
 |NUMERIC|non-reserved|non-reserved|non-reserved|
+|OBJECT|non-reserved|non-reserved|non-reserved|
 |OF|non-reserved|non-reserved|reserved|
 |OFFSET|reserved|non-reserved|reserved|
+|OMIT|non-reserved|non-reserved|reserved|
 |ON|reserved|strict-non-reserved|reserved|
 |ONLY|reserved|non-reserved|reserved|
 |OPEN|non-reserved|non-reserved|reserved|
@@ -714,6 +719,7 @@ Below is a list of all the keywords in Spark SQL.
 |QUALIFY|non-reserved|non-reserved|non-reserved|
 |QUARTER|non-reserved|non-reserved|non-reserved|
 |QUERY|non-reserved|non-reserved|non-reserved|
+|QUOTES|non-reserved|non-reserved|non-reserved|
 |RANGE|non-reserved|non-reserved|reserved|
 |READ|non-reserved|non-reserved|non-reserved|
 |READS|non-reserved|non-reserved|non-reserved|
@@ -824,6 +830,7 @@ Below is a list of all the keywords in Spark SQL.
 |UNARCHIVE|non-reserved|non-reserved|non-reserved|
 |UNBOUNDED|non-reserved|non-reserved|non-reserved|
 |UNCACHE|non-reserved|non-reserved|non-reserved|
+|UNCONDITIONAL|non-reserved|non-reserved|non-reserved|
 |UNIFORM|non-reserved|non-reserved|non-reserved|
 |UNION|reserved|strict-non-reserved|reserved|
 |UNIQUE|reserved|non-reserved|reserved|
@@ -858,6 +865,7 @@ Below is a list of all the keywords in Spark SQL.
 |WITH|reserved|non-reserved|reserved|
 |WITHIN|reserved|non-reserved|reserved|
 |WITHOUT|non-reserved|non-reserved|non-reserved|
+|WRAPPER|non-reserved|non-reserved|non-reserved|
 |X|non-reserved|non-reserved|non-reserved|
 |YEAR|non-reserved|non-reserved|non-reserved|
 |YEARS|non-reserved|non-reserved|non-reserved|
diff --git a/docs/sql-ref-syntax-qry-select-json-query.md b/docs/sql-ref-syntax-qry-select-json-query.md
new file mode 100644
index 000000000000..2cbc60674dfb
--- /dev/null
+++ b/docs/sql-ref-syntax-qry-select-json-query.md
@@ -0,0 +1,174 @@
+---
+layout: global
+title: JSON_QUERY
+displayTitle: JSON_QUERY
+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_QUERY` function extracts the JSON value located by a SQL/JSON path from a JSON document
+and returns it as JSON text (a `STRING`). This is the SQL-standard way (SQL:2016) to pull an object,
+array, or scalar fragment out of JSON, and is commonly used to migrate queries from other systems
+such as Oracle, SQL Server, and Trino. Unlike
+[JSON_TABLE](sql-ref-syntax-qry-select-json-table.html), which produces rows in a `FROM` clause,
+`JSON_QUERY` is an expression that can appear anywhere a value is allowed.
+
+Where [JSON_VALUE](sql-ref-syntax-qry-select-json-value.html) returns a single scalar (and treats an
+object or array match as an error), `JSON_QUERY` returns the matched value serialized as JSON text,
+whether it is an object, an array, or a scalar.
+
+This implementation supports simple, wildcard-free SQL/JSON paths only. The `PASSING` clause, path
+predicates and filters, and explicit `lax` / `strict` path modes defined by SQL:2016 are not
+supported.
+
+### Syntax
+
+```sql
+JSON_QUERY ( json_expr, path
+             [ RETURNING data_type ]
+             [ wrapper_behavior ]
+             [ quotes_behavior ]
+             [ empty_behavior ON EMPTY ]
+             [ error_behavior ON ERROR ] )
+
+wrapper_behavior
+    { WITHOUT [ ARRAY ] WRAPPER
+    | WITH [ CONDITIONAL | UNCONDITIONAL ] [ ARRAY ] WRAPPER }
+
+quotes_behavior
+    { KEEP QUOTES | OMIT QUOTES }
+
+empty_behavior
+    { NULL | ERROR | EMPTY ARRAY | EMPTY OBJECT }
+
+error_behavior
+    { NULL | ERROR | EMPTY ARRAY | EMPTY OBJECT }
+```
+
+### Parameters
+
+* **json_expr**
+
+    An expression that evaluates to a `STRING` containing the JSON document. A `NULL` input yields
+    `NULL` directly (it triggers neither the `ON EMPTY` nor the `ON ERROR` behavior).
+
+* **path**
+
+    A SQL/JSON path literal that locates the value, for example `'$.a.b'` or `'$.items[0]'`. The
+    path must be wildcard-free; a path containing `[*]` is rejected at analysis time.
+
+* **RETURNING data_type**
+
+    The type of the result. It must be a string type; the result is JSON text. If `RETURNING` is
+    omitted, the result type is `STRING`.
+
+* **wrapper_behavior**
+
+    Whether to wrap the result in a JSON array:
+    * `WITHOUT ARRAY WRAPPER` (the default) returns the value unwrapped.
+    * `WITH UNCONDITIONAL ARRAY WRAPPER` (or simply `WITH ARRAY WRAPPER`) always wraps the value in a
+      one-element array.
+    * `WITH CONDITIONAL ARRAY WRAPPER` wraps the value only when it is a scalar; an object or array
+      is returned unwrapped.
+
+* **quotes_behavior**
+
+    Whether to keep the surrounding quotes of a scalar string result:
+    * `KEEP QUOTES` (the default) leaves them, so a string is returned as a quoted JSON string.
+    * `OMIT QUOTES` strips them, returning the raw string content. It is a no-op for objects,
+      arrays, and non-string scalars, and cannot be combined with an array wrapper.
+
+* **empty_behavior ON EMPTY**
+
+    What to produce when `path` matches nothing:
+    * `NULL` (the default) returns SQL `NULL`.
+    * `ERROR` raises an error.
+    * `EMPTY ARRAY` returns the JSON text `[]`.
+    * `EMPTY OBJECT` returns the JSON text `{}`.
+
+* **error_behavior ON ERROR**
+
+    What to produce when the input is not well-formed JSON. The same four choices as `ON EMPTY`
+    apply, defaulting to `NULL`.
+
+A path that matches an explicit JSON `null` is a present scalar value and returns the JSON text
+`null` (it is neither the `ON EMPTY` nor the `ON ERROR` case).
+
+Returning a scalar under the default `WITHOUT ARRAY WRAPPER` is an intentional convenience: the
+matched scalar is emitted as JSON text (for example, `JSON_QUERY('{"id":7}', '$.id')` returns `7`),
+whereas strict SQL:2016 treats a scalar without a wrapper as an error. The wrapper clauses behave the
+standard way: `WITH CONDITIONAL ARRAY WRAPPER` wraps a scalar in a one-element array (`7` becomes
+`[7]`) while leaving a single object or array unwrapped, and `WITH UNCONDITIONAL ARRAY WRAPPER`
+always wraps.
+
+### Examples
+
+```sql
+-- Extract an object as JSON text
+SELECT json_query('{"id":7,"addr":{"city":"NYC"}}', '$.addr');
++---------------------------------------------------+
+|json_query({"id":7,"addr":{"city":"NYC"}}, $.addr) |
++---------------------------------------------------+
+|{"city":"NYC"}                                     |
++---------------------------------------------------+
+
+-- Extract an array
+SELECT json_query('{"tags":["x","y"]}', '$.tags');
++-------------------------------------------+
+|json_query({"tags":["x","y"]}, $.tags)     |
++-------------------------------------------+
+|["x","y"]                                  |
++-------------------------------------------+
+
+-- Wrap a scalar in an array with WITH ARRAY WRAPPER
+-- (WITH ARRAY WRAPPER is a shorthand; the column name shows the canonical
+--  WITH UNCONDITIONAL ARRAY WRAPPER form)
+SELECT json_query('{"tags":["x","y"]}', '$.tags[0]' WITH ARRAY WRAPPER);
++----------------------------------------------------------------------------+
+|json_query({"tags":["x","y"]}, $.tags[0] WITH UNCONDITIONAL ARRAY WRAPPER)  |
++----------------------------------------------------------------------------+
+|["x"]                                                                       |
++----------------------------------------------------------------------------+
+
+-- Strip the quotes from a scalar string with OMIT QUOTES
+SELECT json_query('{"name":"Ada"}', '$.name' OMIT QUOTES);
++---------------------------------------------------+
+|json_query({"name":"Ada"}, $.name OMIT QUOTES)     |
++---------------------------------------------------+
+|Ada                                                |
++---------------------------------------------------+
+
+-- A missing path defaults to NULL; supply a fallback with EMPTY ARRAY ON EMPTY
+SELECT json_query('{"id":7}', '$.missing' EMPTY ARRAY ON EMPTY);
++---------------------------------------------------------+
+|json_query({"id":7}, $.missing EMPTY ARRAY ON EMPTY)     |
++---------------------------------------------------------+
+|[]                                                       |
++---------------------------------------------------------+
+
+-- ERROR ON ERROR raises instead of returning a value
+SELECT json_query('not json', '$.a' ERROR ON ERROR);
+[JSON_QUERY_ON_ERROR.ERROR] ...
+```
+
+### Related Statements
+
+* [SELECT](sql-ref-syntax-qry-select.html)
+* [JSON_VALUE](sql-ref-syntax-qry-select-json-value.html)
+* [JSON_TABLE](sql-ref-syntax-qry-select-json-table.html)
+* [Built-in Functions](sql-ref-functions-builtin.html)
diff --git a/docs/sql-ref-syntax-qry-select-json-value.md b/docs/sql-ref-syntax-qry-select-json-value.md
index deb0be82c2b5..6f45a80417b5 100644
--- a/docs/sql-ref-syntax-qry-select-json-value.md
+++ b/docs/sql-ref-syntax-qry-select-json-value.md
@@ -29,8 +29,10 @@ migrate queries from other systems such as Oracle, DB2, and MySQL. Unlike
 `JSON_VALUE` is an expression that can appear anywhere a scalar is allowed.
 
 The function returns a scalar only. A path that matches an object or array is an *error* case (see
-`ON ERROR`), not a value. Use [JSON_TABLE](sql-ref-syntax-qry-select-json-table.html) or the
-built-in `get_json_object` function to extract structured fragments.
+`ON ERROR`), not a value. To extract an object or array as a JSON fragment, use
+[JSON_QUERY](sql-ref-syntax-qry-select-json-query.html); to produce rows from a JSON array, use
+[JSON_TABLE](sql-ref-syntax-qry-select-json-table.html) (the built-in `get_json_object` function
+also extracts fragments).
 
 ### Syntax
 
diff --git a/docs/sql-ref-syntax-qry-select.md b/docs/sql-ref-syntax-qry-select.md
index b2afd160c5a3..689572d64267 100644
--- a/docs/sql-ref-syntax-qry-select.md
+++ b/docs/sql-ref-syntax-qry-select.md
@@ -213,6 +213,7 @@ SELECT [ hints , ... ] [ ALL | DISTINCT ] { [ [ named_expression | regex_column_
 * [Set Operators](sql-ref-syntax-qry-select-setops.html)
 * [TABLESAMPLE](sql-ref-syntax-qry-select-sampling.html)
 * [Table-valued Function](sql-ref-syntax-qry-select-tvf.html)
+* [JSON_QUERY](sql-ref-syntax-qry-select-json-query.html)
 * [JSON_VALUE](sql-ref-syntax-qry-select-json-value.html)
 * [Window Function](sql-ref-syntax-qry-select-window.html)
 * [CASE Clause](sql-ref-syntax-qry-select-case.html)
diff --git a/docs/sql-ref-syntax.md b/docs/sql-ref-syntax.md
index c8ddecaf6bdb..be4631570bee 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_QUERY](sql-ref-syntax-qry-select-json-query.html)
    * [JSON_TABLE](sql-ref-syntax-qry-select-json-table.html)
    * [JSON_VALUE](sql-ref-syntax-qry-select-json-value.html)
    * [WHERE Clause](sql-ref-syntax-qry-select-where.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 c042bca5e8ce..f82477da3689 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
@@ -201,6 +201,7 @@ COMPENSATION: 'COMPENSATION';
 COMPUTE: 'COMPUTE';
 CONCATENATE: 'CONCATENATE';
 CONDITION: 'CONDITION';
+CONDITIONAL: 'CONDITIONAL';
 CONSTRAINT: 'CONSTRAINT';
 CONTAINS: 'CONTAINS';
 CONTINUE: 'CONTINUE';
@@ -335,8 +336,10 @@ ITEMS: 'ITEMS';
 ITERATE: 'ITERATE';
 JOIN: 'JOIN';
 JSON: 'JSON';
+JSON_QUERY: 'JSON_QUERY';
 JSON_TABLE: 'JSON_TABLE';
 JSON_VALUE: 'JSON_VALUE';
+KEEP: 'KEEP';
 KEY: 'KEY';
 KEYS: 'KEYS';
 LANGUAGE: 'LANGUAGE';
@@ -395,8 +398,10 @@ NULL: 'NULL';
 NULLS: 'NULLS';
 NUMERIC: 'NUMERIC';
 NORELY: 'NORELY';
+OBJECT: 'OBJECT';
 OF: 'OF';
 OFFSET: 'OFFSET';
+OMIT: 'OMIT';
 ON: 'ON';
 ONLY: 'ONLY';
 OPEN: 'OPEN';
@@ -430,6 +435,7 @@ PURGE: 'PURGE';
 QUALIFY: 'QUALIFY';
 QUARTER: 'QUARTER';
 QUERY: 'QUERY';
+QUOTES: 'QUOTES';
 RANGE: 'RANGE';
 READ: 'READ';
 READS: 'READS';
@@ -540,6 +546,7 @@ TYPE: 'TYPE';
 UNARCHIVE: 'UNARCHIVE';
 UNBOUNDED: 'UNBOUNDED';
 UNCACHE: 'UNCACHE';
+UNCONDITIONAL: 'UNCONDITIONAL';
 UNIFORM: 'UNIFORM';
 UNION: 'UNION';
 UNIQUE: 'UNIQUE';
@@ -574,6 +581,7 @@ WINDOW: 'WINDOW';
 WITH: 'WITH';
 WITHIN: 'WITHIN';
 WITHOUT: 'WITHOUT';
+WRAPPER: 'WRAPPER';
 YEAR: 'YEAR';
 YEARS: 'YEARS';
 ZONE: 'ZONE';
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 ee7de7eac2bf..093e89154d5e 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
@@ -1466,6 +1466,12 @@ primaryExpression
       (RETURNING returning=dataType)?
       (emptyBehavior=jsonValueBehavior ON EMPTY)?
       (errorBehavior=jsonValueBehavior ON ERROR)? RIGHT_PAREN                                  #jsonValue
+    | JSON_QUERY LEFT_PAREN jsonExpr=valueExpression COMMA path=stringLit
+      (RETURNING returning=dataType)?
+      wrapper=jsonQueryArrayWrapper?
+      quotes=jsonQueryQuotes?
+      (emptyBehavior=jsonQueryBehavior ON EMPTY)?
+      (errorBehavior=jsonQueryBehavior ON ERROR)? RIGHT_PAREN                                  #jsonQuery
     | constant                                                                                 #constantDefault
     | ASTERISK exceptClause?                                                                   #star
     | qualifiedName DOT ASTERISK exceptClause?                                                 #star
@@ -1500,6 +1506,29 @@ jsonValueBehavior
     | DEFAULT defaultExpr=expression                                                            #jsonValueBehaviorDefault
     ;
 
+// The JSON_QUERY array-wrapper clause. `WITH [UNCONDITIONAL]` always wraps the result in `[...]`;
+// `WITH CONDITIONAL` wraps only a non-array/object (scalar) result; `WITHOUT` (default) never wraps.
+// The `ARRAY` word is optional, matching the SQL standard (`WITH WRAPPER` == `WITH ARRAY WRAPPER`).
+jsonQueryArrayWrapper
+    : WITHOUT ARRAY? WRAPPER                                                                    #jsonQueryWrapperWithout
+    | WITH wrapperType=(CONDITIONAL | UNCONDITIONAL)? ARRAY? WRAPPER                            #jsonQueryWrapperWith
+    ;
+
+// The JSON_QUERY quotes clause: `OMIT QUOTES` strips the surrounding quotes from a scalar string
+// result; `KEEP QUOTES` (default) leaves them.
+jsonQueryQuotes
+    : KEEP QUOTES                                                                              #jsonQueryQuotesKeep
+    | OMIT QUOTES                                                                              #jsonQueryQuotesOmit
+    ;
+
+// The behavior selected by a JSON_QUERY `... ON EMPTY` / `... ON ERROR` clause.
+jsonQueryBehavior
+    : NULL                                                                                     #jsonQueryBehaviorNull
+    | ERROR                                                                                     #jsonQueryBehaviorError
+    | EMPTY ARRAY                                                                              #jsonQueryBehaviorEmptyArray
+    | EMPTY OBJECT                                                                             #jsonQueryBehaviorEmptyObject
+    ;
+
 semiStructuredExtractionPath
     : jsonPathFirstPart (jsonPathParts)*
     ;
@@ -2141,6 +2170,7 @@ ansiNonReserved
     | COMPUTE
     | CONCATENATE
     | CONDITION
+    | CONDITIONAL
     | CONTAINS
     | CONTINUE
     | COST
@@ -2244,8 +2274,10 @@ ansiNonReserved
     | ITEMS
     | ITERATE
     | JSON
+    | JSON_QUERY
     | JSON_TABLE
     | JSON_VALUE
+    | KEEP
     | KEY
     | KEYS
     | LANGUAGE
@@ -2297,7 +2329,9 @@ ansiNonReserved
     | NORELY
     | NULLS
     | NUMERIC
+    | OBJECT
     | OF
+    | OMIT
     | OPEN
     | OPTION
     | OPTIONS
@@ -2324,6 +2358,7 @@ ansiNonReserved
     | QUALIFY
     | QUARTER
     | QUERY
+    | QUOTES
     | RANGE
     | READ
     | READS
@@ -2421,6 +2456,7 @@ ansiNonReserved
     | UNARCHIVE
     | UNBOUNDED
     | UNCACHE
+    | UNCONDITIONAL
     | UNIFORM
     | UNLOCK
     | UNNEST
@@ -2446,6 +2482,7 @@ ansiNonReserved
     | WIDTH
     | WINDOW
     | WITHOUT
+    | WRAPPER
     | YEAR
     | YEARS
     | ZONE
@@ -2555,6 +2592,7 @@ nonReserved
     | COMPUTE
     | CONCATENATE
     | CONDITION
+    | CONDITIONAL
     | CONSTRAINT
     | CONTAINS
     | CONTINUE
@@ -2683,8 +2721,10 @@ nonReserved
     | ITEMS
     | ITERATE
     | JSON
+    | JSON_QUERY
     | JSON_TABLE
     | JSON_VALUE
+    | KEEP
     | KEY
     | KEYS
     | LANGUAGE
@@ -2741,8 +2781,10 @@ nonReserved
     | NULL
     | NULLS
     | NUMERIC
+    | OBJECT
     | OF
     | OFFSET
+    | OMIT
     | ONLY
     | OPEN
     | OPTION
@@ -2775,6 +2817,7 @@ nonReserved
     | QUALIFY
     | QUARTER
     | QUERY
+    | QUOTES
     | RANGE
     | READ
     | READS
@@ -2881,6 +2924,7 @@ nonReserved
     | UNARCHIVE
     | UNBOUNDED
     | UNCACHE
+    | UNCONDITIONAL
     | UNIFORM
     | UNIQUE
     | UNKNOWN
@@ -2913,6 +2957,7 @@ nonReserved
     | WITH
     | WITHIN
     | WITHOUT
+    | WRAPPER
     | YEAR
     | YEARS
     | ZONE
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 bf0891ef7236..9b0ce495a4a5 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
@@ -392,6 +392,31 @@ object JsonValueLookup {
   case class Scalar(text: UTF8String) extends JsonValueLookup
 }
 
+/**
+ * The result of a single-value [[JsonTableEvaluator.queryLookup]] for `JSON_QUERY`. Unlike
+ * [[JsonValueLookup]] -- which reports an object/array match as `NonScalar` without serializing it,
+ * since `JSON_VALUE` never returns a non-scalar -- `Found` here always carries the matched value's
+ * verbatim JSON text (`JSON_QUERY` returns objects, arrays, and scalars alike). `structural`
+ * distinguishes an object/array match from a scalar match, which the caller needs for the array
+ * wrapper (`WITH CONDITIONAL`) and quotes (`OMIT QUOTES`) behaviors. A JSON `null` literal is a
+ * scalar match whose text is `null`, not a distinct case.
+ */
+sealed trait JsonQueryLookup
+object JsonQueryLookup {
+  /** The path did not match (routes to ON EMPTY). */
+  case object Missing extends JsonQueryLookup
+  /**
+   * The path matched a value; `raw` is its verbatim JSON text (a string is still quoted), and
+   * `structural` is true iff the value is an object or array. `unquoted` is the `OMIT QUOTES`
+   * form -- a matched JSON string's decoded content (read straight from the parser), and `raw`
+   * itself for every other value (objects, arrays, numbers, booleans, and JSON `null`, for which
+   * `OMIT QUOTES` is a no-op). Carrying it here lets the caller apply `OMIT QUOTES` without
+   * re-parsing the serialized fragment.
+   */
+  case class Found(raw: UTF8String, structural: Boolean, unquoted: UTF8String)
+    extends JsonQueryLookup
+}
+
 /**
  * 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
@@ -545,6 +570,61 @@ case class JsonTableEvaluator(containerPath: Seq[PathInstruction], explodeRoot:
     }
   }
 
+  /**
+   * Resolves `containerPath` against a single JSON value for `JSON_QUERY`, serializing the matched
+   * value as verbatim JSON text. Returns:
+   *
+   *   - `None` if the input is not a single well-formed JSON value (malformed / trailing garbage /
+   *     empty), which the caller maps to ON ERROR;
+   *   - `Some(Missing)` if the path matches nothing (ON EMPTY);
+   *   - `Some(Found(raw, structural, unquoted))` if the path matches, where `raw` is the value's
+   *     verbatim JSON text, `structural` is true for an object or array (as opposed to a scalar,
+   *     including a JSON `null`, whose text is `null`), and `unquoted` is the `OMIT QUOTES` form
+   *     (a matched JSON string's decoded content; `raw` for every other value).
+   *
+   * A `null` input is the caller's responsibility. Like [[lookup]] this navigates and validates
+   * with a single parser: after the matched value is serialized (which consumes it),
+   * [[drainToRootEnd]]
+   * walks out of the enclosing containers and rejects any trailing content, so a valid prefix
+   * followed by garbage is rejected exactly as a fully malformed document is.
+   */
+  final def queryLookup(json: UTF8String): Option[JsonQueryLookup] = {
+    Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, json)) { parser =>
+      try {
+        if (parser.nextToken() == null) {
+          None // empty or whitespace-only
+        } else {
+          val result = positionAt(parser, containerPath) match {
+            case PositionResult.Missing => JsonQueryLookup.Missing
+            // A JSON `null` literal is a scalar value for JSON_QUERY: serialize it to the text
+            // `null` rather than reporting it specially. The parser is positioned on the token.
+            case PositionResult.NullValue =>
+              val raw = serializeCurrentValue(parser)
+              JsonQueryLookup.Found(raw, structural = false, unquoted = raw)
+            case PositionResult.AtValue =>
+              parser.currentToken match {
+                case JsonToken.START_OBJECT | JsonToken.START_ARRAY =>
+                  val raw = serializeCurrentValue(parser)
+                  JsonQueryLookup.Found(raw, structural = true, unquoted = raw)
+                case JsonToken.VALUE_STRING =>
+                  // Capture the decoded string straight from the parser so `OMIT QUOTES` need not
+                  // re-parse the serialized (re-quoted) form.
+                  val unquoted = UTF8String.fromString(parser.getText)
+                  JsonQueryLookup.Found(serializeCurrentValue(parser), structural = false, unquoted)
+                case _ =>
+                  // A non-string scalar (number/boolean): `OMIT QUOTES` is a no-op.
+                  val raw = serializeCurrentValue(parser)
+                  JsonQueryLookup.Found(raw, structural = false, unquoted = raw)
+              }
+          }
+          if (drainToRootEnd(parser)) Some(result) else None
+        }
+      } catch {
+        case _: JsonProcessingException => None
+      }
+    }
+  }
+
   /**
    * Classifies the value the parser is positioned at (the `AtValue` case of [[positionAt]]) for a
    * `JSON_VALUE` [[lookup]], avoiding the serialize-then-reparse round trip that
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 c91221690091..218ea2e1d28d 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
@@ -24,9 +24,9 @@ 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, JsonPathResult, JsonTableEvaluator, JsonTablePathTrie, JsonToStructsEvaluator,
-  JsonTupleEvaluator, JsonValueLookup, MultiGetJsonObjectEvaluator, PathInstruction,
-  SchemaOfJsonEvaluator, StructsToJsonEvaluator}
+  JsonPathParser, JsonPathResult, JsonQueryLookup, JsonTableEvaluator, JsonTablePathTrie,
+  JsonToStructsEvaluator, JsonTupleEvaluator, JsonValueLookup, 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,
@@ -859,6 +859,222 @@ object JsonValue {
   }
 }
 
+/**
+ * Behavior of `JSON_QUERY`'s `ON EMPTY` / `ON ERROR` clause: what to produce when the path matches
+ * nothing (`ON EMPTY`) or the input is not valid JSON (`ON ERROR`).
+ */
+sealed trait JsonQueryBehavior
+object JsonQueryBehavior {
+  /** Produce SQL NULL (the SQL-standard default for both clauses). */
+  case object Null extends JsonQueryBehavior
+  /** Raise an error. */
+  case object Error extends JsonQueryBehavior
+  /** Produce an empty JSON array `[]`. */
+  case object EmptyArray extends JsonQueryBehavior
+  /** Produce an empty JSON object `{}`. */
+  case object EmptyObject extends JsonQueryBehavior
+}
+
+/**
+ * The array-wrapper behavior of `JSON_QUERY` (SQL:2016 `... ARRAY WRAPPER`). This implementation
+ * resolves a single value per path (wildcard-free paths only), so a wrapper wraps that value in a
+ * one-element array:
+ *   - `Without` (default): return the value unwrapped;
+ *   - `Unconditional` (`WITH [UNCONDITIONAL] ARRAY WRAPPER`): always wrap;
+ *   - `Conditional` (`WITH CONDITIONAL ARRAY WRAPPER`): wrap only a scalar; leave an object or
+ *     array as is.
+ */
+sealed trait JsonQueryWrapper
+object JsonQueryWrapper {
+  case object Without extends JsonQueryWrapper
+  case object Conditional extends JsonQueryWrapper
+  case object Unconditional extends JsonQueryWrapper
+}
+
+/** The quotes behavior of `JSON_QUERY`: `KEEP QUOTES` (default) or `OMIT QUOTES`. */
+sealed trait JsonQueryQuotes
+object JsonQueryQuotes {
+  case object Keep extends JsonQueryQuotes
+  case object Omit extends JsonQueryQuotes
+}
+
+// scalastyle:off line.size.limit
+/**
+ * The SQL:2016 `JSON_QUERY` function (feature T828): extracts the JSON value located by `path` from
+ * a JSON input and returns it as JSON text (STRING):
+ *
+ *   - missing path                        -> ON EMPTY behavior
+ *   - malformed / non-single-value input  -> ON ERROR behavior
+ *   - matched object / array / scalar     -> its verbatim JSON text, after applying the array
+ *                                            wrapper and quotes clauses
+ *
+ * A matched scalar (including a JSON `null`) is not an error under the default `WITHOUT ARRAY
+ * WRAPPER`; it is emitted as JSON text (`JSON_QUERY('{"id":7}', '$.id')` -> `7`). `OMIT QUOTES`
+ * strips the surrounding quotes from a scalar string result (and cannot be combined with a wrapper).
+ * Both `ON EMPTY` and `ON ERROR` default to NULL per the standard, and a `null` JSON input yields
+ * SQL NULL directly. `RETURNING` is restricted to string types here (VARIANT is deferred); the
+ * result is always JSON text.
+ *
+ * {{{
+ *   JSON_QUERY('{"a":{"x":1}}', '$.a')                          -- '{"x":1}'
+ *   JSON_QUERY('{"t":["x","y"]}', '$.t')                        -- '["x","y"]'
+ *   JSON_QUERY('{"t":["x","y"]}', '$.t[0]' WITH ARRAY WRAPPER)  -- '["x"]'
+ *   JSON_QUERY('{"n":"Ada"}', '$.n' OMIT QUOTES)                -- 'Ada'
+ * }}}
+ */
+// scalastyle:on line.size.limit
+case class JsonQuery(
+    child: Expression,
+    path: String,
+    returning: DataType,
+    wrapper: JsonQueryWrapper,
+    quotes: JsonQueryQuotes,
+    onEmpty: JsonQueryBehavior,
+    onError: JsonQueryBehavior)
+  extends UnaryExpression
+  with CodegenFallback
+  with ExpectsInputTypes
+  with QueryErrorsBase {
+
+  override def nullable: Boolean = true
+
+  // The JSON input must be a STRING; the result is JSON text.
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(StringTypeWithCollation(supportsTrimCollation = true))
+
+  override def dataType: DataType = returning
+
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val inputCheck = super.checkInputDataTypes()
+    if (inputCheck.isFailure) {
+      inputCheck
+    } else if (!JsonPathParser.hasWildcard(path).contains(false)) {
+      // The path must parse and be wildcard-free (a single value is resolved).
+      DataTypeMismatch(
+        errorSubClass = "INVALID_JSON_PATH",
+        messageParameters = Map(
+          "functionName" -> toSQLId(prettyName), "path" -> toSQLValue(path)))
+    } else if (!JsonQuery.isValidReturningType(returning)) {
+      // RETURNING is restricted to string types (the result is JSON text; VARIANT is deferred).
+      DataTypeMismatch(
+        errorSubClass = "INVALID_JSON_QUERY_RETURNING_TYPE",
+        messageParameters = Map(
+          "functionName" -> toSQLId(prettyName), "returningType" -> toSQLType(returning)))
+    } else if (quotes == JsonQueryQuotes.Omit && wrapper != JsonQueryWrapper.Without) {
+      // OMIT QUOTES applies only to an unwrapped scalar; the SQL standard forbids pairing it with
+      // an array wrapper. Enforced here (not only in the parser) so a directly-constructed
+      // expression cannot silently ignore the quotes clause.
+      DataTypeMismatch(
+        errorSubClass = "INVALID_JSON_QUERY_WRAPPER_AND_QUOTES",
+        messageParameters = Map("functionName" -> toSQLId(prettyName)))
+    } else {
+      TypeCheckResult.TypeCheckSuccess
+    }
+  }
+
+  // Path parsed once (the grammar makes it a string literal). `checkInputDataTypes` guarantees it
+  // parses and is wildcard-free, so the evaluator is only built for a valid path.
+  @transient private lazy val evaluator: JsonTableEvaluator =
+    JsonTableEvaluator(JsonPathParser.parse(path).getOrElse(Nil), explodeRoot = false)
+
+  // Handle the ON EMPTY / ON ERROR case per the configured behavior.
+  private def onEmptyResult(): Any = behaviorResult(onEmpty, isEmpty = true)
+  private def onErrorResult(): Any = behaviorResult(onError, isEmpty = false)
+
+  private def behaviorResult(behavior: JsonQueryBehavior, isEmpty: Boolean): Any = behavior match {
+    case JsonQueryBehavior.Null => null
+    case JsonQueryBehavior.EmptyArray => JsonQuery.EmptyArrayText
+    case JsonQueryBehavior.EmptyObject => JsonQuery.EmptyObjectText
+    case JsonQueryBehavior.Error =>
+      if (isEmpty) throw QueryExecutionErrors.jsonQueryOnEmptyError(prettyName, path, cause = null)
+      else throw QueryExecutionErrors.jsonQueryOnErrorError(prettyName, path, cause = null)
+  }
+
+  // Apply the array-wrapper and quotes clauses to a matched value. `raw` is its verbatim JSON text,
+  // `unquoted` is the OMIT QUOTES form (a string's decoded content; `raw` otherwise, so OMIT QUOTES
+  // is a no-op for objects, arrays, and non-string scalars), and `structural` is true for an object
+  // or array match (rather than a scalar, incl. JSON null).
+  private def wrapAndQuote(raw: UTF8String, unquoted: UTF8String, structural: Boolean): UTF8String =
+    wrapper match {
+      case JsonQueryWrapper.Without =>
+        // OMIT QUOTES reuses the string decoded during the lookup rather than re-parsing the
+        // serialized fragment; OMIT QUOTES combined with a wrapper is rejected at parse time.
+        if (quotes == JsonQueryQuotes.Omit) unquoted else raw
+      case JsonQueryWrapper.Unconditional => JsonQuery.wrapInArray(raw)
+      // CONDITIONAL wraps only a scalar; an object or array is already a structural result.
+      case JsonQueryWrapper.Conditional => if (structural) raw else JsonQuery.wrapInArray(raw)
+    }
+
+  override def eval(input: InternalRow): Any = {
+    val json = child.eval(input).asInstanceOf[UTF8String]
+    // NULL input propagates to NULL (not ON EMPTY / ON ERROR), matching ANSI and the other engines.
+    if (json == null) return null
+    evaluator.queryLookup(json) match {
+      // Malformed / non-single-value input.
+      case None => onErrorResult()
+      // Path matched nothing.
+      case Some(JsonQueryLookup.Missing) => onEmptyResult()
+      // Matched a value: serialize it, applying the wrapper and quotes clauses.
+      case Some(JsonQueryLookup.Found(raw, structural, unquoted)) =>
+        wrapAndQuote(raw, unquoted, structural)
+    }
+  }
+
+  override def prettyName: String = "json_query"
+
+  override def sql: String = {
+    val returningSQL = if (returning == StringType) "" else s" RETURNING ${returning.sql}"
+    val wrapperSQL = wrapper match {
+      case JsonQueryWrapper.Without => ""
+      case JsonQueryWrapper.Unconditional => " WITH UNCONDITIONAL ARRAY WRAPPER"
+      case JsonQueryWrapper.Conditional => " WITH CONDITIONAL ARRAY WRAPPER"
+    }
+    val quotesSQL = quotes match {
+      case JsonQueryQuotes.Keep => ""
+      case JsonQueryQuotes.Omit => " OMIT QUOTES"
+    }
+    def behaviorSQL(b: JsonQueryBehavior): String = b match {
+      case JsonQueryBehavior.Null => "NULL"
+      case JsonQueryBehavior.Error => "ERROR"
+      case JsonQueryBehavior.EmptyArray => "EMPTY ARRAY"
+      case JsonQueryBehavior.EmptyObject => "EMPTY OBJECT"
+    }
+    val emptySQL =
+      if (onEmpty == JsonQueryBehavior.Null) "" else s" ${behaviorSQL(onEmpty)} ON EMPTY"
+    val errorSQL =
+      if (onError == JsonQueryBehavior.Null) "" else s" ${behaviorSQL(onError)} ON ERROR"
+    // Render the path as a properly escaped string literal so bracket-quoted paths round-trip.
+    val pathSQL = Literal(UTF8String.fromString(path), StringType).sql
+    s"JSON_QUERY(${child.sql}, $pathSQL$returningSQL$wrapperSQL$quotesSQL$emptySQL$errorSQL)"
+  }
+
+  override protected def withNewChildInternal(newChild: Expression): JsonQuery =
+    copy(child = newChild)
+}
+
+object JsonQuery {
+  private val EmptyArrayText: UTF8String = UTF8String.fromString("[]")
+  private val EmptyObjectText: UTF8String = UTF8String.fromString("{}")
+  private val ArrayOpen: UTF8String = UTF8String.fromString("[")
+  private val ArrayClose: UTF8String = UTF8String.fromString("]")
+
+  private def wrapInArray(raw: UTF8String): UTF8String =
+    UTF8String.concat(ArrayOpen, raw, ArrayClose)
+
+  /**
+   * `JSON_QUERY` returns a JSON fragment as text, so RETURNING is restricted to a plain STRING here
+   * (VARIANT is deferred). `CharType` / `VarcharType` extend `StringType` but carry a length that
+   * `JSON_QUERY` does not enforce -- it returns the fragment verbatim without a cast -- so they are
+   * rejected: the parser normalizes a SQL `CHAR`/`VARCHAR` RETURNING to STRING before construction,
+   * and this guards a raw `CharType`/`VarcharType` supplied by direct Catalyst construction.
+   */
+  def isValidReturningType(dt: DataType): Boolean = dt match {
+    case _: CharType | _: VarcharType => false
+    case _: StringType => true
+    case _ => false
+  }
+}
+
 /**
  * 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 74c691699609..6f46bcccea81 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
@@ -4200,6 +4200,58 @@ class AstBuilder extends DataTypeAstBuilder
     JsonValue(jsonExpr, path, returning, onEmpty, onError, emptyDefault, errorDefault)
   }
 
+  /**
+   * Resolve a `jsonQueryBehavior` clause (`NULL` / `ERROR` / `EMPTY ARRAY` / `EMPTY OBJECT`) into a
+   * [[JsonQueryBehavior]].
+   */
+  private def buildJsonQueryBehavior(ctx: JsonQueryBehaviorContext): JsonQueryBehavior = ctx match {
+    case _: JsonQueryBehaviorNullContext => JsonQueryBehavior.Null
+    case _: JsonQueryBehaviorErrorContext => JsonQueryBehavior.Error
+    case _: JsonQueryBehaviorEmptyArrayContext => JsonQueryBehavior.EmptyArray
+    case _: JsonQueryBehaviorEmptyObjectContext => JsonQueryBehavior.EmptyObject
+  }
+
+  /**
+   * Create a [[JsonQuery]] expression for the SQL:2016 `JSON_QUERY` function. The array wrapper
+   * defaults to `WITHOUT ARRAY WRAPPER`, quotes to `KEEP QUOTES`, and both `ON EMPTY` / `ON ERROR`
+   * to `NULL`, per the standard. `OMIT QUOTES` cannot be combined with an array wrapper.
+   */
+  override def visitJsonQuery(ctx: JsonQueryContext): Expression = withOrigin(ctx) {
+    val jsonExpr = expression(ctx.jsonExpr)
+    val path = string(visitStringLit(ctx.path))
+    // Default RETURNING type is STRING; the result is JSON text. A CHAR/VARCHAR RETURNING is
+    // normalized to STRING truly unconditionally: JSON_QUERY returns the fragment verbatim without
+    // a length-enforcing cast, so the result type must never advertise a CHAR/VARCHAR length it
+    // cannot enforce. The CharVarcharUtils helpers cannot be used here: they honor
+    // spark.sql.preserveCharVarcharTypeInfo and would leave a VARCHAR(n) length in the output type
+    // when that flag is set. A non-string RETURNING is left intact for checkInputDataTypes to fail.
+    val returning = Option(ctx.returning).map(typedVisit[DataType]).map {
+      case c: CharType => c.toStringType
+      case v: VarcharType => v.toStringType
+      case other => other
+    }.getOrElse(StringType)
+    val wrapper = Option(ctx.wrapper).map {
+      case _: JsonQueryWrapperWithoutContext => JsonQueryWrapper.Without
+      case w: JsonQueryWrapperWithContext =>
+        if (w.wrapperType != null && w.wrapperType.getType == SqlBaseParser.CONDITIONAL) {
+          JsonQueryWrapper.Conditional
+        } else {
+          JsonQueryWrapper.Unconditional
+        }
+    }.getOrElse(JsonQueryWrapper.Without)
+    val quotes = Option(ctx.quotes).map {
+      case _: JsonQueryQuotesKeepContext => JsonQueryQuotes.Keep
+      case _: JsonQueryQuotesOmitContext => JsonQueryQuotes.Omit
+    }.getOrElse(JsonQueryQuotes.Keep)
+    // The OMIT QUOTES + array-wrapper invariant is enforced in JsonQuery.checkInputDataTypes so it
+    // holds for directly-constructed expressions too, not only this parser path.
+    val onEmpty =
+      Option(ctx.emptyBehavior).map(buildJsonQueryBehavior).getOrElse(JsonQueryBehavior.Null)
+    val onError =
+      Option(ctx.errorBehavior).map(buildJsonQueryBehavior).getOrElse(JsonQueryBehavior.Null)
+    JsonQuery(jsonExpr, path, returning, wrapper, quotes, onEmpty, onError)
+  }
+
   /**
    * Create a (windowed) Function expression.
    */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 646d296c11c6..5fd7745bd3a9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -1625,6 +1625,20 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       cause = e)
   }
 
+  def jsonQueryOnEmptyError(functionName: String, path: String, cause: Throwable): Throwable = {
+    new SparkRuntimeException(
+      errorClass = "JSON_QUERY_ON_ERROR.EMPTY",
+      messageParameters = Map("functionName" -> toSQLId(functionName), "path" -> toSQLValue(path)),
+      cause = cause)
+  }
+
+  def jsonQueryOnErrorError(functionName: String, path: String, cause: Throwable): Throwable = {
+    new SparkRuntimeException(
+      errorClass = "JSON_QUERY_ON_ERROR.ERROR",
+      messageParameters = Map("functionName" -> toSQLId(functionName), "path" -> toSQLValue(path)),
+      cause = cause)
+  }
+
   def jsonValueOnEmptyError(functionName: String, path: String, cause: Throwable): Throwable = {
     new SparkRuntimeException(
       errorClass = "JSON_VALUE_ON_ERROR.EMPTY",
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala
index 18a92a6496d3..47aadbdcef6e 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala
@@ -331,6 +331,51 @@ class ExpressionParserSuite extends AnalysisTest {
         Some(Literal("x")), Some(Literal("y"))))
   }
 
+  test("JSON_QUERY expressions") {
+    import org.apache.spark.sql.catalyst.expressions.{JsonQueryBehavior, JsonQueryQuotes,
+      JsonQueryWrapper}
+    // Bare form: default STRING RETURNING, WITHOUT wrapper, KEEP quotes, NULL ON EMPTY / ON ERROR.
+    assertEqual(
+      "json_query(a, '$.b')",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Without, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    // WITH ARRAY WRAPPER defaults to UNCONDITIONAL; the ARRAY word is optional.
+    assertEqual(
+      "json_query(a, '$.b' WITH ARRAY WRAPPER)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Unconditional, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    assertEqual(
+      "json_query(a, '$.b' WITH CONDITIONAL WRAPPER)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Conditional, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    // WITHOUT ARRAY WRAPPER with OMIT QUOTES.
+    assertEqual(
+      "json_query(a, '$.b' WITHOUT ARRAY WRAPPER OMIT QUOTES)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Without, JsonQueryQuotes.Omit,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    // EMPTY ARRAY ON EMPTY / EMPTY OBJECT ON ERROR, plus RETURNING STRING.
+    assertEqual(
+      "json_query(a, '$.b' RETURNING STRING EMPTY ARRAY ON EMPTY EMPTY OBJECT ON ERROR)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Without, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.EmptyArray, JsonQueryBehavior.EmptyObject))
+    // The ARRAY word is optional in every wrapper spelling, and WITH alone means UNCONDITIONAL.
+    Seq("WITH WRAPPER", "WITH UNCONDITIONAL WRAPPER").foreach { spelling =>
+      assertEqual(
+        s"json_query(a, '$$.b' $spelling)",
+        JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Unconditional, JsonQueryQuotes.Keep,
+          JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    }
+    assertEqual(
+      "json_query(a, '$.b' WITHOUT WRAPPER)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Without, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+    // Explicit NULL ON EMPTY / NULL ON ERROR (same as the omitted default) and KEEP QUOTES.
+    assertEqual(
+      "json_query(a, '$.b' KEEP QUOTES NULL ON EMPTY NULL ON ERROR)",
+      JsonQuery($"a", "$.b", StringType, JsonQueryWrapper.Without, JsonQueryQuotes.Keep,
+        JsonQueryBehavior.Null, JsonQueryBehavior.Null))
+  }
+
   test("cast expressions") {
     // Note that DataType parsing is tested elsewhere.
     assertEqual("cast(a as int)", $"a".cast(IntegerType))
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 b07f9da119a4..7bd9611d1d38 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,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,EMPTY,ENFORCED,ERROR,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HISTORY,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,JSON_TABLE,JSON_VALUE,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,ORDINALITY,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURNING,ROLE,ROLES,SCD,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRACK,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,WIDTH,X,YEARS,ZONE")
+      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,CONDITIONAL,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,EMPTY,ENFORCED,ERROR,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HISTORY,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,JSON_QUERY,JSON_TABLE,JSON_VALUE,KEEP,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OBJECT,OFFSET,OMIT,OPTION,OPTIONS,ORDINALITY,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,QUOTES,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURNING,ROLE,ROLES,SCD,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRACK,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNCONDITIONAL,UNIFORM,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,WIDTH,WRAPPER,X,YEARS,ZONE")
       // scalastyle:on line.size.limit
     }
   }
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/json-functions.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/json-functions.sql.out
index 9b33afd10db8..6a9e2db602f7 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/json-functions.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/json-functions.sql.out
@@ -1358,3 +1358,222 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "json_value('{}', '$.x' RETURNING INT DEFAULT array(1) ON EMPTY)"
   } ]
 }
+
+
+-- !query
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.addr')
+-- !query analysis
+Project [json_query({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, $.addr, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, '$.addr')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.tags')
+-- !query analysis
+Project [json_query({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, $.tags, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, '$.tags')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.id')
+-- !query analysis
+Project [json_query({"id":7}, $.id, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"id":7}, '$.id')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name')
+-- !query analysis
+Project [json_query({"name":"Ada"}, $.name, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"name":"Ada"}, '$.name')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"score":null}', '$.score')
+-- !query analysis
+Project [json_query({"score":null}, $.score, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"score":null}, '$.score')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.missing')
+-- !query analysis
+Project [json_query({"id":7}, $.missing, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"id":7}, '$.missing')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query(cast(null as string), '$.a')
+-- !query analysis
+Project [json_query(cast(null as string), $.a, StringType, Without, Keep, Null, Null) AS JSON_QUERY(CAST(NULL AS STRING), '$.a')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"tags":["x","y"]}', '$.tags[0]' WITH ARRAY WRAPPER)
+-- !query analysis
+Project [json_query({"tags":["x","y"]}, $.tags[0], StringType, Unconditional, Keep, Null, Null) AS JSON_QUERY({"tags":["x","y"]}, '$.tags[0]' WITH UNCONDITIONAL ARRAY WRAPPER)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"tags":["x","y"]}', '$.tags' WITH UNCONDITIONAL ARRAY WRAPPER)
+-- !query analysis
+Project [json_query({"tags":["x","y"]}, $.tags, StringType, Unconditional, Keep, Null, Null) AS JSON_QUERY({"tags":["x","y"]}, '$.tags' WITH UNCONDITIONAL ARRAY WRAPPER)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.id' WITH ARRAY WRAPPER)
+-- !query analysis
+Project [json_query({"id":7}, $.id, StringType, Unconditional, Keep, Null, Null) AS JSON_QUERY({"id":7}, '$.id' WITH UNCONDITIONAL ARRAY WRAPPER)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.id' WITH CONDITIONAL ARRAY WRAPPER)
+-- !query analysis
+Project [json_query({"id":7}, $.id, StringType, Conditional, Keep, Null, Null) AS JSON_QUERY({"id":7}, '$.id' WITH CONDITIONAL ARRAY WRAPPER)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' WITH CONDITIONAL ARRAY WRAPPER)
+-- !query analysis
+Project [json_query({"addr":{"city":"NYC"}}, $.addr, StringType, Conditional, Keep, Null, Null) AS JSON_QUERY({"addr":{"city":"NYC"}}, '$.addr' WITH CONDITIONAL ARRAY WRAPPER)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' OMIT QUOTES)
+-- !query analysis
+Project [json_query({"name":"Ada"}, $.name, StringType, Without, Omit, Null, Null) AS JSON_QUERY({"name":"Ada"}, '$.name' OMIT QUOTES)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' KEEP QUOTES)
+-- !query analysis
+Project [json_query({"name":"Ada"}, $.name, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"name":"Ada"}, '$.name')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' EMPTY ARRAY ON EMPTY)
+-- !query analysis
+Project [json_query({"id":7}, $.missing, StringType, Without, Keep, EmptyArray, Null) AS JSON_QUERY({"id":7}, '$.missing' EMPTY ARRAY ON EMPTY)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' EMPTY OBJECT ON EMPTY)
+-- !query analysis
+Project [json_query({"id":7}, $.missing, StringType, Without, Keep, EmptyObject, Null) AS JSON_QUERY({"id":7}, '$.missing' EMPTY OBJECT ON EMPTY)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' ERROR ON EMPTY)
+-- !query analysis
+Project [json_query({"id":7}, $.missing, StringType, Without, Keep, Error, Null) AS JSON_QUERY({"id":7}, '$.missing' ERROR ON EMPTY)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('not json', '$.a')
+-- !query analysis
+Project [json_query(not json, $.a, StringType, Without, Keep, Null, Null) AS JSON_QUERY(not json, '$.a')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('not json', '$.a' EMPTY ARRAY ON ERROR)
+-- !query analysis
+Project [json_query(not json, $.a, StringType, Without, Keep, Null, EmptyArray) AS JSON_QUERY(not json, '$.a' EMPTY ARRAY ON ERROR)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('not json', '$.a' EMPTY OBJECT ON ERROR)
+-- !query analysis
+Project [json_query(not json, $.a, StringType, Without, Keep, Null, EmptyObject) AS JSON_QUERY(not json, '$.a' EMPTY OBJECT ON ERROR)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('not json', '$.a' ERROR ON ERROR)
+-- !query analysis
+Project [json_query(not json, $.a, StringType, Without, Keep, Null, Error) AS JSON_QUERY(not json, '$.a' ERROR ON ERROR)#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' RETURNING STRING)
+-- !query analysis
+Project [json_query({"addr":{"city":"NYC"}}, $.addr, StringType, Without, Keep, Null, Null) AS JSON_QUERY({"addr":{"city":"NYC"}}, '$.addr')#x]
++- OneRowRelation
+
+
+-- !query
+select json_query('{"a":[1,2]}', '$.a[*]')
+-- !query analysis
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_PATH",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "path" : "'$.a[*]'",
+    "sqlExpr" : "\"JSON_QUERY({\"a\":[1,2]}, '$.a[*]')\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 42,
+    "fragment" : "json_query('{\"a\":[1,2]}', '$.a[*]')"
+  } ]
+}
+
+
+-- !query
+select json_query('{"a":1}', '$.a' RETURNING INT)
+-- !query analysis
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_QUERY_RETURNING_TYPE",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "returningType" : "\"INT\"",
+    "sqlExpr" : "\"JSON_QUERY({\"a\":1}, '$.a' RETURNING INT)\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 49,
+    "fragment" : "json_query('{\"a\":1}', '$.a' RETURNING INT)"
+  } ]
+}
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' WITH ARRAY WRAPPER OMIT QUOTES)
+-- !query analysis
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_QUERY_WRAPPER_AND_QUOTES",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "sqlExpr" : "\"JSON_QUERY({\"name\":\"Ada\"}, '$.name' WITH UNCONDITIONAL ARRAY WRAPPER OMIT QUOTES)\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 76,
+    "fragment" : "json_query('{\"name\":\"Ada\"}', '$.name' WITH ARRAY WRAPPER OMIT QUOTES)"
+  } ]
+}
diff --git a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
index fa1cb6628ec9..2af052c29134 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
@@ -211,3 +211,43 @@ select json_value('{"a":[1,2]}', '$.a[*]');
 select json_value('{"a":1}', '$.a' RETURNING STRUCT<x:INT>);
 -- invalid: a DEFAULT that cannot cast to the RETURNING type
 select json_value('{}', '$.x' RETURNING INT DEFAULT array(1) ON EMPTY);
+
+-- JSON_QUERY: extract an object or array as JSON text (ANSI SQL:2016)
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.addr');
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.tags');
+-- a scalar result is emitted as JSON text (not an error) under the default WITHOUT ARRAY WRAPPER
+select json_query('{"id":7}', '$.id');
+select json_query('{"name":"Ada"}', '$.name');
+-- present but JSON null -> the JSON text null
+select json_query('{"score":null}', '$.score');
+-- missing path -> NULL ON EMPTY (default)
+select json_query('{"id":7}', '$.missing');
+-- NULL input propagates to NULL
+select json_query(cast(null as string), '$.a');
+-- ARRAY WRAPPER
+select json_query('{"tags":["x","y"]}', '$.tags[0]' WITH ARRAY WRAPPER);
+select json_query('{"tags":["x","y"]}', '$.tags' WITH UNCONDITIONAL ARRAY WRAPPER);
+select json_query('{"id":7}', '$.id' WITH ARRAY WRAPPER);
+-- CONDITIONAL wraps only a scalar; an object/array is left as is
+select json_query('{"id":7}', '$.id' WITH CONDITIONAL ARRAY WRAPPER);
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' WITH CONDITIONAL ARRAY WRAPPER);
+-- OMIT QUOTES strips the quotes from a scalar string result
+select json_query('{"name":"Ada"}', '$.name' OMIT QUOTES);
+select json_query('{"name":"Ada"}', '$.name' KEEP QUOTES);
+-- ON EMPTY behaviors
+select json_query('{"id":7}', '$.missing' EMPTY ARRAY ON EMPTY);
+select json_query('{"id":7}', '$.missing' EMPTY OBJECT ON EMPTY);
+select json_query('{"id":7}', '$.missing' ERROR ON EMPTY);
+-- ON ERROR behaviors (malformed input)
+select json_query('not json', '$.a');
+select json_query('not json', '$.a' EMPTY ARRAY ON ERROR);
+select json_query('not json', '$.a' EMPTY OBJECT ON ERROR);
+select json_query('not json', '$.a' ERROR ON ERROR);
+-- RETURNING STRING is allowed (the result is JSON text)
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' RETURNING STRING);
+-- invalid: wildcard path
+select json_query('{"a":[1,2]}', '$.a[*]');
+-- invalid: non-string RETURNING type
+select json_query('{"a":1}', '$.a' RETURNING INT);
+-- invalid: OMIT QUOTES combined with an array wrapper
+select json_query('{"name":"Ada"}', '$.name' WITH ARRAY WRAPPER OMIT QUOTES);
diff --git a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
index 6656a7723cd1..15f34b5633eb 100644
--- a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
@@ -1546,3 +1546,266 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
     "fragment" : "json_value('{}', '$.x' RETURNING INT DEFAULT array(1) ON EMPTY)"
   } ]
 }
+
+
+-- !query
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.addr')
+-- !query schema
+struct<JSON_QUERY({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, '$.addr'):string>
+-- !query output
+{"city":"NYC"}
+
+
+-- !query
+select json_query('{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}', '$.tags')
+-- !query schema
+struct<JSON_QUERY({"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}, '$.tags'):string>
+-- !query output
+["x","y"]
+
+
+-- !query
+select json_query('{"id":7}', '$.id')
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.id'):string>
+-- !query output
+7
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name')
+-- !query schema
+struct<JSON_QUERY({"name":"Ada"}, '$.name'):string>
+-- !query output
+"Ada"
+
+
+-- !query
+select json_query('{"score":null}', '$.score')
+-- !query schema
+struct<JSON_QUERY({"score":null}, '$.score'):string>
+-- !query output
+null
+
+
+-- !query
+select json_query('{"id":7}', '$.missing')
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.missing'):string>
+-- !query output
+NULL
+
+
+-- !query
+select json_query(cast(null as string), '$.a')
+-- !query schema
+struct<JSON_QUERY(CAST(NULL AS STRING), '$.a'):string>
+-- !query output
+NULL
+
+
+-- !query
+select json_query('{"tags":["x","y"]}', '$.tags[0]' WITH ARRAY WRAPPER)
+-- !query schema
+struct<JSON_QUERY({"tags":["x","y"]}, '$.tags[0]' WITH UNCONDITIONAL ARRAY WRAPPER):string>
+-- !query output
+["x"]
+
+
+-- !query
+select json_query('{"tags":["x","y"]}', '$.tags' WITH UNCONDITIONAL ARRAY WRAPPER)
+-- !query schema
+struct<JSON_QUERY({"tags":["x","y"]}, '$.tags' WITH UNCONDITIONAL ARRAY WRAPPER):string>
+-- !query output
+[["x","y"]]
+
+
+-- !query
+select json_query('{"id":7}', '$.id' WITH ARRAY WRAPPER)
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.id' WITH UNCONDITIONAL ARRAY WRAPPER):string>
+-- !query output
+[7]
+
+
+-- !query
+select json_query('{"id":7}', '$.id' WITH CONDITIONAL ARRAY WRAPPER)
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.id' WITH CONDITIONAL ARRAY WRAPPER):string>
+-- !query output
+[7]
+
+
+-- !query
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' WITH CONDITIONAL ARRAY WRAPPER)
+-- !query schema
+struct<JSON_QUERY({"addr":{"city":"NYC"}}, '$.addr' WITH CONDITIONAL ARRAY WRAPPER):string>
+-- !query output
+{"city":"NYC"}
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' OMIT QUOTES)
+-- !query schema
+struct<JSON_QUERY({"name":"Ada"}, '$.name' OMIT QUOTES):string>
+-- !query output
+Ada
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' KEEP QUOTES)
+-- !query schema
+struct<JSON_QUERY({"name":"Ada"}, '$.name'):string>
+-- !query output
+"Ada"
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' EMPTY ARRAY ON EMPTY)
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.missing' EMPTY ARRAY ON EMPTY):string>
+-- !query output
+[]
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' EMPTY OBJECT ON EMPTY)
+-- !query schema
+struct<JSON_QUERY({"id":7}, '$.missing' EMPTY OBJECT ON EMPTY):string>
+-- !query output
+{}
+
+
+-- !query
+select json_query('{"id":7}', '$.missing' ERROR ON EMPTY)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkRuntimeException
+{
+  "errorClass" : "JSON_QUERY_ON_ERROR.EMPTY",
+  "sqlState" : "2203G",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "path" : "'$.missing'"
+  }
+}
+
+
+-- !query
+select json_query('not json', '$.a')
+-- !query schema
+struct<JSON_QUERY(not json, '$.a'):string>
+-- !query output
+NULL
+
+
+-- !query
+select json_query('not json', '$.a' EMPTY ARRAY ON ERROR)
+-- !query schema
+struct<JSON_QUERY(not json, '$.a' EMPTY ARRAY ON ERROR):string>
+-- !query output
+[]
+
+
+-- !query
+select json_query('not json', '$.a' EMPTY OBJECT ON ERROR)
+-- !query schema
+struct<JSON_QUERY(not json, '$.a' EMPTY OBJECT ON ERROR):string>
+-- !query output
+{}
+
+
+-- !query
+select json_query('not json', '$.a' ERROR ON ERROR)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkRuntimeException
+{
+  "errorClass" : "JSON_QUERY_ON_ERROR.ERROR",
+  "sqlState" : "2203G",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "path" : "'$.a'"
+  }
+}
+
+
+-- !query
+select json_query('{"addr":{"city":"NYC"}}', '$.addr' RETURNING STRING)
+-- !query schema
+struct<JSON_QUERY({"addr":{"city":"NYC"}}, '$.addr'):string>
+-- !query output
+{"city":"NYC"}
+
+
+-- !query
+select json_query('{"a":[1,2]}', '$.a[*]')
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_PATH",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "path" : "'$.a[*]'",
+    "sqlExpr" : "\"JSON_QUERY({\"a\":[1,2]}, '$.a[*]')\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 42,
+    "fragment" : "json_query('{\"a\":[1,2]}', '$.a[*]')"
+  } ]
+}
+
+
+-- !query
+select json_query('{"a":1}', '$.a' RETURNING INT)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_QUERY_RETURNING_TYPE",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "returningType" : "\"INT\"",
+    "sqlExpr" : "\"JSON_QUERY({\"a\":1}, '$.a' RETURNING INT)\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 49,
+    "fragment" : "json_query('{\"a\":1}', '$.a' RETURNING INT)"
+  } ]
+}
+
+
+-- !query
+select json_query('{"name":"Ada"}', '$.name' WITH ARRAY WRAPPER OMIT QUOTES)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "DATATYPE_MISMATCH.INVALID_JSON_QUERY_WRAPPER_AND_QUOTES",
+  "sqlState" : "42K09",
+  "messageParameters" : {
+    "functionName" : "`json_query`",
+    "sqlExpr" : "\"JSON_QUERY({\"name\":\"Ada\"}, '$.name' WITH UNCONDITIONAL ARRAY WRAPPER OMIT QUOTES)\""
+  },
+  "queryContext" : [ {
+    "objectType" : "",
+    "objectName" : "",
+    "startIndex" : 8,
+    "stopIndex" : 76,
+    "fragment" : "json_query('{\"name\":\"Ada\"}', '$.name' WITH ARRAY WRAPPER OMIT QUOTES)"
+  } ]
+}
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 37f174acfe53..d748acb49227 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
@@ -77,6 +77,7 @@ COMPENSATION	false
 COMPUTE	false
 CONCATENATE	false
 CONDITION	false
+CONDITIONAL	false
 CONSTRAINT	true
 CONTAINS	false
 CONTINUE	false
@@ -212,8 +213,10 @@ ITEMS	false
 ITERATE	false
 JOIN	true
 JSON	false
+JSON_QUERY	false
 JSON_TABLE	false
 JSON_VALUE	false
+KEEP	false
 KEY	false
 KEYS	false
 LANGUAGE	false
@@ -272,8 +275,10 @@ NOT	true
 NULL	true
 NULLS	false
 NUMERIC	false
+OBJECT	false
 OF	false
 OFFSET	true
+OMIT	false
 ON	true
 ONLY	true
 OPEN	false
@@ -307,6 +312,7 @@ PURGE	false
 QUALIFY	false
 QUARTER	false
 QUERY	false
+QUOTES	false
 RANGE	false
 READ	false
 READS	false
@@ -413,6 +419,7 @@ TYPE	false
 UNARCHIVE	false
 UNBOUNDED	false
 UNCACHE	false
+UNCONDITIONAL	false
 UNIFORM	false
 UNION	true
 UNIQUE	true
@@ -447,6 +454,7 @@ WINDOW	false
 WITH	true
 WITHIN	true
 WITHOUT	false
+WRAPPER	false
 X	false
 YEAR	false
 YEARS	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 f6b910a54c3a..dac6326238da 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
@@ -77,6 +77,7 @@ COMPENSATION	false
 COMPUTE	false
 CONCATENATE	false
 CONDITION	false
+CONDITIONAL	false
 CONSTRAINT	false
 CONTAINS	false
 CONTINUE	false
@@ -212,8 +213,10 @@ ITEMS	false
 ITERATE	false
 JOIN	false
 JSON	false
+JSON_QUERY	false
 JSON_TABLE	false
 JSON_VALUE	false
+KEEP	false
 KEY	false
 KEYS	false
 LANGUAGE	false
@@ -272,8 +275,10 @@ NOT	false
 NULL	false
 NULLS	false
 NUMERIC	false
+OBJECT	false
 OF	false
 OFFSET	false
+OMIT	false
 ON	false
 ONLY	false
 OPEN	false
@@ -307,6 +312,7 @@ PURGE	false
 QUALIFY	false
 QUARTER	false
 QUERY	false
+QUOTES	false
 RANGE	false
 READ	false
 READS	false
@@ -413,6 +419,7 @@ TYPE	false
 UNARCHIVE	false
 UNBOUNDED	false
 UNCACHE	false
+UNCONDITIONAL	false
 UNIFORM	false
 UNION	false
 UNIQUE	false
@@ -447,6 +454,7 @@ WINDOW	false
 WITH	false
 WITHIN	false
 WITHOUT	false
+WRAPPER	false
 X	false
 YEAR	false
 YEARS	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 f6b910a54c3a..dac6326238da 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
@@ -77,6 +77,7 @@ COMPENSATION	false
 COMPUTE	false
 CONCATENATE	false
 CONDITION	false
+CONDITIONAL	false
 CONSTRAINT	false
 CONTAINS	false
 CONTINUE	false
@@ -212,8 +213,10 @@ ITEMS	false
 ITERATE	false
 JOIN	false
 JSON	false
+JSON_QUERY	false
 JSON_TABLE	false
 JSON_VALUE	false
+KEEP	false
 KEY	false
 KEYS	false
 LANGUAGE	false
@@ -272,8 +275,10 @@ NOT	false
 NULL	false
 NULLS	false
 NUMERIC	false
+OBJECT	false
 OF	false
 OFFSET	false
+OMIT	false
 ON	false
 ONLY	false
 OPEN	false
@@ -307,6 +312,7 @@ PURGE	false
 QUALIFY	false
 QUARTER	false
 QUERY	false
+QUOTES	false
 RANGE	false
 READ	false
 READS	false
@@ -413,6 +419,7 @@ TYPE	false
 UNARCHIVE	false
 UNBOUNDED	false
 UNCACHE	false
+UNCONDITIONAL	false
 UNIFORM	false
 UNION	false
 UNIQUE	false
@@ -447,6 +454,7 @@ WINDOW	false
 WITH	false
 WITHIN	false
 WITHOUT	false
+WRAPPER	false
 X	false
 YEAR	false
 YEARS	false
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonQuerySuite.scala
new file mode 100644
index 000000000000..083d319f6c27
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonQuerySuite.scala
@@ -0,0 +1,228 @@
+/*
+ * 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.SparkRuntimeException
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
+import org.apache.spark.sql.catalyst.expressions.{JsonQuery, JsonQueryBehavior, JsonQueryQuotes,
+  JsonQueryWrapper, Literal}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{CharType, StringType, VarcharType}
+
+/**
+ * End-to-end tests for the SQL:2016 `JSON_QUERY` function.
+ */
+class JsonQuerySuite extends QueryTest with SharedSparkSession {
+  import testImplicits._
+
+  private val doc =
+    """{"id":7,"name":"Ada","tags":["x","y"],"addr":{"city":"NYC"},"score":null}"""
+
+  test("extract an object or array as verbatim JSON text") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.addr')"), Row("""{"city":"NYC"}"""))
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.tags')"), Row("""["x","y"]"""))
+  }
+
+  test("default RETURNING type is STRING") {
+    assert(sql(s"SELECT json_query('$doc', '$$.addr')").schema.head.dataType === StringType)
+  }
+
+  test("RETURNING VARCHAR/CHAR is normalized to STRING and does not truncate") {
+    // JSON_QUERY returns the fragment verbatim (no length-enforcing cast), so a CHAR/VARCHAR
+    // RETURNING must not advertise a length it cannot enforce. The result type is STRING and the
+    // value is not truncated -- including when char/varchar type info is otherwise preserved.
+    Seq("false", "true").foreach { preserve =>
+      withSQLConf(SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> preserve) {
+        // VARCHAR(n) and CHAR(n) exercise the two separate normalization branches.
+        Seq("VARCHAR(2)", "CHAR(2)").foreach { returning =>
+          val df = sql(s"SELECT json_query('$doc', '$$.addr' RETURNING $returning)")
+          assert(df.schema.head.dataType === StringType, s"$returning preserve=$preserve")
+          checkAnswer(df, Row("""{"city":"NYC"}"""))
+        }
+      }
+    }
+  }
+
+  test("a directly-constructed JsonQuery with a CHAR/VARCHAR RETURNING is rejected") {
+    // The parser normalizes CHAR/VARCHAR to STRING, but a raw CharType/VarcharType supplied by
+    // direct Catalyst construction would otherwise advertise a length JSON_QUERY does not enforce.
+    // isValidReturningType rejects it, so checkInputDataTypes fails.
+    Seq(VarcharType(2), CharType(2)).foreach { returning =>
+      val expr = JsonQuery(Literal("{}"), "$.a", returning, JsonQueryWrapper.Without,
+        JsonQueryQuotes.Keep, JsonQueryBehavior.Null, JsonQueryBehavior.Null)
+      expr.checkInputDataTypes() match {
+        case DataTypeMismatch(errorSubClass, _) =>
+          assert(errorSubClass == "INVALID_JSON_QUERY_RETURNING_TYPE", s"for $returning")
+        case other => fail(s"expected DataTypeMismatch for $returning, got $other")
+      }
+    }
+  }
+
+  test("RETURNING STRING is allowed (the result is JSON text)") {
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.addr' RETURNING STRING)"), Row("""{"city":"NYC"}"""))
+  }
+
+  test("a scalar result is emitted as JSON text under the default WITHOUT ARRAY WRAPPER") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.id')"), Row("7"))
+    // A string scalar keeps its surrounding quotes by default (KEEP QUOTES).
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.name')"), Row("\"Ada\""))
+  }
+
+  test("a present JSON null yields the JSON text null") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.score')"), Row("null"))
+  }
+
+  test("a missing path is an ON EMPTY case, NULL by default") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.missing')"), Row(null))
+  }
+
+  test("NULL JSON input propagates to NULL (not ON EMPTY / ON ERROR)") {
+    checkAnswer(sql("SELECT json_query(CAST(NULL AS STRING), '$.a')"), Row(null))
+  }
+
+  test("WITH [UNCONDITIONAL] ARRAY WRAPPER wraps the result in a one-element array") {
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.tags[0]' WITH ARRAY WRAPPER)"), Row("""["x"]"""))
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.tags' WITH UNCONDITIONAL ARRAY WRAPPER)"),
+      Row("""[["x","y"]]"""))
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.id' WITH ARRAY WRAPPER)"), Row("[7]"))
+  }
+
+  test("WITH CONDITIONAL ARRAY WRAPPER wraps only a scalar") {
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.id' WITH CONDITIONAL ARRAY WRAPPER)"), Row("[7]"))
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.addr' WITH CONDITIONAL ARRAY WRAPPER)"),
+      Row("""{"city":"NYC"}"""))
+    checkAnswer(
+      sql(s"SELECT json_query('$doc', '$$.tags' WITH CONDITIONAL ARRAY WRAPPER)"),
+      Row("""["x","y"]"""))
+  }
+
+  test("OMIT QUOTES strips the quotes from a scalar string result") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.name' OMIT QUOTES)"), Row("Ada"))
+    // KEEP QUOTES is the default and keeps them.
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.name' KEEP QUOTES)"), Row("\"Ada\""))
+    // OMIT QUOTES is a no-op for a non-string scalar and for structural results.
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.id' OMIT QUOTES)"), Row("7"))
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.addr' OMIT QUOTES)"), Row("""{"city":"NYC"}"""))
+  }
+
+  test("OMIT QUOTES unescapes an escaped string scalar") {
+    // Pass the JSON via a column so SQL string-literal escaping does not rewrite it first. The
+    // stored JSON is {"s":"a\"b\\c\n"}; s decodes to a, quote, b, backslash, c, newline.
+    val df = Seq("""{"s":"a\"b\\c\n"}""").toDF("j")
+    // KEEP QUOTES (default) returns the verbatim, re-escaped JSON string.
+    checkAnswer(df.selectExpr("json_query(j, '$.s')"), Row(""""a\"b\\c\n""""))
+    // OMIT QUOTES returns the raw unescaped content, which is intentionally no longer valid JSON.
+    checkAnswer(df.selectExpr("json_query(j, '$.s' OMIT QUOTES)"), Row("a\"b\\c\n"))
+  }
+
+  test("the JSON_QUERY keyword is non-reserved and usable as an identifier") {
+    // JSON_QUERY and the OBJECT keyword introduced for the ON EMPTY / ON ERROR clause are
+    // non-reserved in both modes, so they remain usable as column names.
+    withTable("t") {
+      sql("CREATE TABLE t (json_query INT, object STRING) USING parquet")
+      sql("INSERT INTO t VALUES (1, 'x')")
+      checkAnswer(sql("SELECT json_query, object FROM t"), Row(1, "x"))
+    }
+  }
+
+  test("EMPTY ARRAY / EMPTY OBJECT ON EMPTY") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.missing' EMPTY ARRAY ON EMPTY)"), Row("[]"))
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.missing' EMPTY OBJECT ON EMPTY)"), Row("{}"))
+  }
+
+  test("ERROR ON EMPTY raises for a missing path") {
+    val e = intercept[SparkRuntimeException] {
+      sql(s"SELECT json_query('$doc', '$$.missing' ERROR ON EMPTY)").collect()
+    }
+    assert(e.getCondition == "JSON_QUERY_ON_ERROR.EMPTY")
+  }
+
+  test("malformed input is an ON ERROR case, NULL by default") {
+    checkAnswer(sql("SELECT json_query('not json', '$.a')"), Row(null))
+    checkAnswer(sql("SELECT json_query('not json', '$.a' EMPTY ARRAY ON ERROR)"), Row("[]"))
+    checkAnswer(sql("SELECT json_query('not json', '$.a' EMPTY OBJECT ON ERROR)"), Row("{}"))
+  }
+
+  test("ERROR ON ERROR raises for malformed input") {
+    val e = intercept[SparkRuntimeException] {
+      sql("SELECT json_query('not json', '$.a' ERROR ON ERROR)").collect()
+    }
+    assert(e.getCondition == "JSON_QUERY_ON_ERROR.ERROR")
+  }
+
+  test("a valid JSON prefix followed by trailing content is an ON ERROR case") {
+    checkAnswer(sql("""SELECT json_query('{"a":{"b":1}} trailing', '$.a')"""), Row(null))
+    checkAnswer(sql("""SELECT json_query('{"a":1}{"a":2}', '$.a')"""), Row(null))
+    val e = intercept[SparkRuntimeException] {
+      sql("""SELECT json_query('{"a":{"b":1}} trailing', '$.a' ERROR ON ERROR)""").collect()
+    }
+    assert(e.getCondition == "JSON_QUERY_ON_ERROR.ERROR")
+  }
+
+  test("nested path into an object and array index") {
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.addr.city')"), Row("\"NYC\""))
+    checkAnswer(sql(s"SELECT json_query('$doc', '$$.tags[1]')"), Row("\"y\""))
+  }
+
+  test("invalid: wildcard path is rejected at analysis") {
+    val e = intercept[AnalysisException] {
+      sql(s"SELECT json_query('$doc', '$$.tags[*]')").collect()
+    }
+    assert(e.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_PATH")
+  }
+
+  test("invalid: non-string RETURNING type is rejected at analysis") {
+    val e = intercept[AnalysisException] {
+      sql(s"SELECT json_query('$doc', '$$.id' RETURNING INT)").collect()
+    }
+    assert(e.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_QUERY_RETURNING_TYPE")
+  }
+
+  test("invalid: OMIT QUOTES combined with an array wrapper is rejected at analysis") {
+    val e = intercept[AnalysisException] {
+      sql(s"SELECT json_query('$doc', '$$.name' WITH ARRAY WRAPPER OMIT QUOTES)").collect()
+    }
+    assert(e.getCondition == "DATATYPE_MISMATCH.INVALID_JSON_QUERY_WRAPPER_AND_QUOTES")
+  }
+
+  test("sql renders a bracket-quoted path as a valid, re-parseable string literal") {
+    val df = sql(s"SELECT json_query('$doc', '$$[\\'addr\\']')")
+    val jsonQuery = df.queryExecution.analyzed.expressions
+      .flatMap(_.collect { case jq: JsonQuery => jq }).head
+    val rendered = jsonQuery.sql
+    assert(rendered.contains("\\'addr\\'"), s"path was not escaped in: $rendered")
+    checkAnswer(sql(s"SELECT $rendered"), Row("""{"city":"NYC"}"""))
+  }
+
+  test("works over a column of JSON documents") {
+    val df = Seq(
+      """{"a":{"x":1}}""",
+      """{"a":[1,2]}""",
+      """{"b":2}""",
+      "not json").toDF("j")
+    checkAnswer(
+      df.selectExpr("json_query(j, '$.a')"),
+      Seq(Row("""{"x":1}"""), Row("[1,2]"), Row(null), Row(null)))
+  }
+}
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 e6aa18256c90..bfa0d7fac064 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,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,EMPTY,END,ENFORCED,ERROR,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HISTORY,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,JSON_TABLE,JSON_VALUE,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,ORDINALITY,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNING,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCD,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRACK,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNNEST,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE")
+      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,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONDITIONAL,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,EMPTY,END,ENFORCED,ERROR,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HISTORY,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,JSON_QUERY,JSON_TABLE,JSON_VALUE,KEEP,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OBJECT,OF,OFFSET,OMIT,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,ORDINALITY,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,QUOTES,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNING,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCD,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRACK,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNCONDITIONAL,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNNEST,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,WRAPPER,X,YEAR,YEARS,ZONE")
       // scalastyle:on line.size.limit
     }
   }