This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b6f2fa017bfb [SPARK-58035][SQL] Add `try_variant_insert` expression
b6f2fa017bfb is described below
commit b6f2fa017bfb668a2f2f2aaa94a1849485640d94
Author: bojana-db <[email protected]>
AuthorDate: Sat Jul 11 17:11:56 2026 +0800
[SPARK-58035][SQL] Add `try_variant_insert` expression
### What changes were proposed in this pull request?
Adds the SQL function `try_variant_insert(v, path, val)`, the
error-tolerant counterpart of `variant_insert` (SPARK-57641). It inserts a
value into a Variant value at a single JSONPath location, returning NULL
instead of failing the query when the insertion hits a recoverable error.
Details:
- On a valid insertion, behaves exactly like `variant_insert`: an object
path (e.g.
`$.b`) adds a new field; an array path (e.g. `$[N]`) inserts at index
`N`, shifting
later elements right and padding gaps past the end with variant nulls;
missing
intermediate keys/indices along the path are created;
- Recoverable errors are caught and return NULL instead of throwing: a
duplicate key
(`VARIANT_DUPLICATE_KEY`) and any path type mismatch
(`VARIANT_PATH_TYPE_MISMATCH`,
e.g. an object key applied to an array);
- Non-recoverable errors still throw, matching `variant_insert`: a
malformed path is
rejected with `INVALID_VARIANT_PATH` (the path is parsed before the
insert), and a
result exceeding the size limit raises `VARIANT_SIZE_LIMIT`;
- Any NULL argument returns NULL;
- The value may be any expression castable to variant.
### Why are the changes needed?
Error-tolerant counterpart of `variant_insert`, matching `try_variant_get`
and other `try_*` functions.
### Does this PR introduce _any_ user-facing change?
Yes, a new SQL function `try_variant_insert` (and the corresponding
Scala/Python
`functions` API).
### How was this patch tested?
Unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor with Claude Opus 4.8
Closes #57120 from bojana-db/try-variant-insert.
Authored-by: bojana-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../source/reference/pyspark.sql/functions.rst | 1 +
python/pyspark/sql/connect/functions/builtin.py | 10 ++
python/pyspark/sql/functions/__init__.py | 1 +
python/pyspark/sql/functions/builtin.py | 63 +++++++-
python/pyspark/sql/tests/test_functions.py | 6 +
.../scala/org/apache/spark/sql/functions.scala | 46 +++++-
.../sql/catalyst/analysis/FunctionRegistry.scala | 3 +-
.../variant/VariantExpressionEvalUtils.scala | 17 +-
.../expressions/variant/variantExpressions.scala | 179 ++++++++++++++-------
.../variant/VariantExpressionSuite.scala | 130 +++++++++------
.../apache/spark/sql/PlanGenerationTestSuite.scala | 4 +
.../function_try_variant_insert.explain | 2 +
.../function_variant_insert.explain | 2 +-
.../queries/function_try_variant_insert.json | 125 ++++++++++++++
.../queries/function_try_variant_insert.proto.bin | Bin 0 -> 1151 bytes
.../sql-functions/sql-expression-schema.md | 3 +-
.../scala/org/apache/spark/sql/VariantSuite.scala | 55 +++++++
17 files changed, 532 insertions(+), 115 deletions(-)
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst
b/python/docs/source/reference/pyspark.sql/functions.rst
index 70e5811180d0..9ae6cd12a13c 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -605,6 +605,7 @@ VARIANT Functions
variant_delete
variant_get
variant_insert
+ try_variant_insert
try_parse_json
to_variant_object
diff --git a/python/pyspark/sql/connect/functions/builtin.py
b/python/pyspark/sql/connect/functions/builtin.py
index 8e74aeef2828..8a33d3b2b38e 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -2227,6 +2227,16 @@ def variant_insert(v: "ColumnOrName", path:
Union[Column, str], value: "ColumnOr
variant_insert.__doc__ = pysparkfuncs.variant_insert.__doc__
+def try_variant_insert(
+ v: "ColumnOrName", path: Union[Column, str], value: "ColumnOrName"
+) -> Column:
+ path_col = path if isinstance(path, Column) else lit(path)
+ return _invoke_function("try_variant_insert", _to_col(v), path_col,
_to_col(value))
+
+
+try_variant_insert.__doc__ = pysparkfuncs.try_variant_insert.__doc__
+
+
def variant_get(v: "ColumnOrName", path: Union[Column, str], targetType: str)
-> Column:
assert isinstance(path, (Column, str))
if isinstance(path, str):
diff --git a/python/pyspark/sql/functions/__init__.py
b/python/pyspark/sql/functions/__init__.py
index ee9d06fcaa2d..052ca8a51737 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -483,6 +483,7 @@ __all__ = [ # noqa: F405
"variant_delete",
"variant_get",
"variant_insert",
+ "try_variant_insert",
"try_parse_json",
"to_variant_object",
# XML Functions
diff --git a/python/pyspark/sql/functions/builtin.py
b/python/pyspark/sql/functions/builtin.py
index 1e348cfd4287..2103822179d6 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -21731,7 +21731,8 @@ def variant_insert(v: "ColumnOrName", path:
Union[Column, str], value: "ColumnOr
"""
Inserts a value into a variant at the given JSONPath location. An object
path adds a new field
(error if it already exists); an array path inserts at the index, shifting
later elements
- right. Missing intermediate keys are created. Returns NULL if any argument
is NULL.
+ right. Missing intermediate keys are created. Throws an error if a path
segment hits a value
+ of an incompatible type. Returns NULL if any argument is NULL.
.. versionadded:: 4.3.0
@@ -21786,6 +21787,66 @@ def variant_insert(v: "ColumnOrName", path:
Union[Column, str], value: "ColumnOr
)
+@_try_remote_functions
+def try_variant_insert(
+ v: "ColumnOrName", path: Union[Column, str], value: "ColumnOrName"
+) -> Column:
+ """
+ Inserts a value into a variant at the given JSONPath location. An object
path adds a new field;
+ an array path inserts at the index, shifting later elements right. Missing
intermediate keys
+ are created. Returns NULL if the field already exists or a path segment
hits a value of an
+ incompatible type, or if any argument is NULL.
+
+ .. versionadded:: 4.3.0
+
+ Parameters
+ ----------
+ v : :class:`~pyspark.sql.Column` or str
+ a variant column or column name
+ path : :class:`~pyspark.sql.Column` or str
+ the JSONPath insertion target. A `str` is a literal path; a
+ :class:`~pyspark.sql.Column` supplies the path at runtime. A valid
path should start with
+ `$` and is followed by one or more segments like `[123]`, `.name`,
`['name']`, or
+ `["name"]`. The root path `$` is not allowed.
+ value : :class:`~pyspark.sql.Column` or str
+ the value to insert. Any expression castable to variant.
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ a variant column with `value` inserted at `path`, or NULL if the
insertion fails
+
+ Examples
+ --------
+ >>> from pyspark.sql.functions import lit, parse_json, to_json,
try_variant_insert
+ >>> df = spark.createDataFrame([{'json': '''{ "a": 1, "arr": ["x", "y"]
}'''}])
+ >>> v = parse_json(df.json)
+ >>> df.select(to_json(try_variant_insert(v, "$.b",
lit(2))).alias("r")).collect()
+ [Row(r='{"a":1,"arr":["x","y"],"b":2}')]
+ >>> df.select(to_json(try_variant_insert(v, "$.c.d",
lit(3))).alias("r")).collect()
+ [Row(r='{"a":1,"arr":["x","y"],"c":{"d":3}}')]
+ >>> df.select(to_json(try_variant_insert(v, "$.arr[1]",
lit("z"))).alias("r")).collect()
+ [Row(r='{"a":1,"arr":["x","z","y"]}')]
+ >>> df.select(to_json(try_variant_insert(v, "$.arr[5]",
lit("z"))).alias("r")).collect()
+ [Row(r='{"a":1,"arr":["x","y",null,null,null,"z"]}')]
+ >>> df.select(to_json(try_variant_insert(v, "$.a",
lit(2))).alias("r")).collect()
+ [Row(r=None)]
+ >>> df.select(to_json(try_variant_insert(v, "$.a.b",
lit(2))).alias("r")).collect()
+ [Row(r=None)]
+ >>> df.select(to_json(try_variant_insert(v, "$.b",
lit(None))).alias("r")).collect()
+ [Row(r=None)]
+ """
+ from pyspark.sql.classic.column import _to_java_column
+
+ path_col = path if isinstance(path, Column) else lit(path)
+ return _invoke_function(
+ "try_variant_insert",
+ _to_java_column(v),
+ _to_java_column(path_col),
+ _to_java_column(value),
+ )
+
+
@_try_remote_functions
def variant_get(v: "ColumnOrName", path: Union[Column, str], targetType: str)
-> Column:
"""
diff --git a/python/pyspark/sql/tests/test_functions.py
b/python/pyspark/sql/tests/test_functions.py
index 2ade286763de..3ac206d71f92 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -3534,6 +3534,12 @@ class FunctionsTestsMixin:
df.select(F.to_json(F.variant_insert(v, df.newpath, F.lit(9)))),
['{"a":1,"z":9}', '{"b":2,"z":9}'],
)
+ check(
+ df.select(F.to_json(F.try_variant_insert(v, "$.z", F.lit(9)))),
+ ['{"a":1,"z":9}', '{"b":2,"z":9}'],
+ )
+ check(df.select(F.to_json(F.try_variant_insert(v, df.path,
F.lit(9)))), [None, None])
+ check(df.select(F.to_json(F.try_variant_insert(v, "$.z",
F.lit(None)))), [None, None])
check(df.select(F.schema_of_variant(v)), ["OBJECT<a: BIGINT>",
"OBJECT<b: BIGINT>"])
check(df.select(F.schema_of_variant_agg(v)), ["OBJECT<a: BIGINT, b:
BIGINT>"])
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
index c8b269a63de5..0be5b0b30c01 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
@@ -9824,7 +9824,8 @@ object functions {
/**
* Inserts a value into a variant at the given JSONPath location. An object
path adds a new
* field (error if it already exists); an array path inserts at the index,
shifting later
- * elements right. Missing intermediate keys are created. Returns NULL if
any argument is NULL.
+ * elements right. Missing intermediate keys are created. Throws an error if
a path segment hits
+ * a value of an incompatible type. Returns NULL if any argument is NULL.
*
* @param v
* a variant column.
@@ -9843,7 +9844,8 @@ object functions {
/**
* Inserts a value into a variant at the given JSONPath location. An object
path adds a new
* field (error if it already exists); an array path inserts at the index,
shifting later
- * elements right. Missing intermediate keys are created. Returns NULL if
any argument is NULL.
+ * elements right. Missing intermediate keys are created. Throws an error if
a path segment hits
+ * a value of an incompatible type. Returns NULL if any argument is NULL.
*
* @param v
* a variant column.
@@ -9859,6 +9861,46 @@ object functions {
def variant_insert(v: Column, path: String, value: Column): Column =
Column.fn("variant_insert", v, lit(path), value)
+ /**
+ * Inserts a value into a variant at the given JSONPath location. An object
path adds a new
+ * field; an array path inserts at the index, shifting later elements right.
Missing
+ * intermediate keys are created. Returns NULL if the field already exists
or a path segment
+ * hits a value of an incompatible type, or if any argument is NULL.
+ *
+ * @param v
+ * a variant column.
+ * @param path
+ * the column containing the JSONPath string identifying the insertion
target. A valid path
+ * should start with `$` and is followed by one or more segments like
`[123]`, `.name`,
+ * `['name']`, or `["name"]`. The root path `$` is not allowed.
+ * @param value
+ * the value to insert. Any expression castable to variant.
+ * @group variant_funcs
+ * @since 4.3.0
+ */
+ def try_variant_insert(v: Column, path: Column, value: Column): Column =
+ Column.fn("try_variant_insert", v, path, value)
+
+ /**
+ * Inserts a value into a variant at the given JSONPath location. An object
path adds a new
+ * field; an array path inserts at the index, shifting later elements right.
Missing
+ * intermediate keys are created. Returns NULL if the field already exists
or a path segment
+ * hits a value of an incompatible type, or if any argument is NULL.
+ *
+ * @param v
+ * a variant column.
+ * @param path
+ * the JSONPath identifying the insertion target. A valid path should
start with `$` and is
+ * followed by one or more segments like `[123]`, `.name`, `['name']`, or
`["name"]`. The root
+ * path `$` is not allowed.
+ * @param value
+ * the value to insert. Any expression castable to variant.
+ * @group variant_funcs
+ * @since 4.3.0
+ */
+ def try_variant_insert(v: Column, path: String, value: Column): Column =
+ Column.fn("try_variant_insert", v, lit(path), value)
+
/**
* Extracts a sub-variant from `v` according to `path` string, and then cast
the sub-variant to
* `targetType`. Returns null if the path does not exist. Throws an
exception if the cast fails.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index 193e939889db..f1b788eb3e81 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -986,7 +986,8 @@ object FunctionRegistry {
expression[ToVariantObject]("to_variant_object"),
expression[IsValidVariant]("is_valid_variant"),
expression[VariantDelete]("variant_delete"),
- expression[VariantInsert]("variant_insert"),
+ expressionBuilder("variant_insert", VariantInsertExpressionBuilder),
+ expressionBuilder("try_variant_insert", TryVariantInsertExpressionBuilder),
// Spatial
expression[ST_AsBinary]("st_asbinary"),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala
index 20a55bccc779..bf5d22ee7482 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions.variant
import scala.util.control.NonFatal
+import org.apache.spark.SparkRuntimeException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -131,7 +132,10 @@ object VariantExpressionEvalUtils {
/**
* Insert `value` into `input` at `javaSegments`. `path` is the source
string used in error
* messages. The cast and insert share one try, so any size overflow maps to
`VARIANT_SIZE_LIMIT`
- * and a type mismatch maps to `VARIANT_PATH_TYPE_MISMATCH`.
+ * and a type mismatch maps to `VARIANT_PATH_TYPE_MISMATCH`. When
`failOnError` is false (the
+ * `try_variant_insert` mode), a duplicate key or path type mismatch returns
null instead of
+ * throwing; a size overflow (and a malformed path, rejected earlier during
parsing) is still
+ * raised.
*/
def insertAtPath(
input: VariantVal,
@@ -139,7 +143,8 @@ object VariantExpressionEvalUtils {
path: String,
value: Any,
valueDataType: DataType,
- functionName: String): VariantVal = {
+ functionName: String,
+ failOnError: Boolean): VariantVal = {
val v = new Variant(input.getValue, input.getMetadata)
try {
val valVal = castToVariant(value, valueDataType)
@@ -147,6 +152,9 @@ object VariantExpressionEvalUtils {
val out = VariantBuilder.insertAtPath(v, javaSegments, valVariant)
new VariantVal(out.getValue, out.getMetadata)
} catch {
+ case _: VariantPathTypeMismatchException if !failOnError => null
+ case e: SparkRuntimeException if !failOnError && e.getCondition ==
"VARIANT_DUPLICATE_KEY" =>
+ null
case e: VariantPathTypeMismatchException =>
throw QueryExecutionErrors.variantPathTypeMismatch(
path, renderVariantPath(javaSegments.take(e.depth)), functionName)
@@ -160,10 +168,11 @@ object VariantExpressionEvalUtils {
path: UTF8String,
value: Any,
valueDataType: DataType,
- functionName: String): VariantVal = {
+ functionName: String,
+ failOnError: Boolean): VariantVal = {
val pathStr = path.toString
val javaSegments = toJavaSegments(parseVariantPath(pathStr, functionName))
- insertAtPath(input, javaSegments, pathStr, value, valueDataType,
functionName)
+ insertAtPath(input, javaSegments, pathStr, value, valueDataType,
functionName, failOnError)
}
/** Cast a Spark value from `dataType` into the variant type. */
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala
index 0f8995ce829f..3e404cc9c190 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala
@@ -808,42 +808,11 @@ object VariantDelete {
}
}
-// scalastyle:off line.size.limit
-@ExpressionDescription(
- usage = "_FUNC_(v, path, val) - Inserts a value into a variant at the given
JSONPath " +
- "location. An object path adds a new field (error if it already exists);
an array path " +
- "inserts at the index, shifting later elements right. Missing intermediate
keys are " +
- "created. Returns NULL if any argument is NULL.",
- arguments = """
- Arguments:
- * v - A variant value to mutate.
- * path - A string expression evaluating to a JSONPath identifying the
insertion target. A
- valid path should start with `$` and is followed by one or more
segments like `[123]`,
- `.name`, `['name']`, or `["name"]`. The root path `$` is not allowed.
- * val - Any expression castable to variant.
- """,
- examples = """
- Examples:
- > SELECT _FUNC_(parse_json('{"a": 1}'), '$.b', 2);
- {"a":1,"b":2}
- > SELECT _FUNC_(parse_json('{}'), '$.a.b', 1);
- {"a":{"b":1}}
- > SELECT _FUNC_(parse_json('["a","b","c"]'), '$[1]', 'z');
- ["a","z","b","c"]
- > SELECT _FUNC_(parse_json('["a","b"]'), '$[5]', 'z');
- ["a","b",null,null,null,"z"]
- > SELECT _FUNC_(parse_json('{}'), '$.a', parse_json('{"x":1}'));
- {"a":{"x":1}}
- > SELECT _FUNC_(parse_json('{}'), '$.a', parse_json('null'));
- {"a":null}
- > SELECT _FUNC_(parse_json('{}'), '$.a', null);
- NULL
- """,
- since = "4.3.0",
- group = "variant_funcs"
-)
-// scalastyle:on line.size.limit
-case class VariantInsert(input: Expression, path: Expression, value:
Expression)
+case class VariantInsert(
+ input: Expression,
+ path: Expression,
+ value: Expression,
+ failOnError: Boolean = true)
extends TernaryExpression
with ExpectsInputTypes
with QueryErrorsBase {
@@ -853,6 +822,7 @@ case class VariantInsert(input: Expression, path:
Expression, value: Expression)
override def third: Expression = value
override def nullIntolerant: Boolean = true
+ override def nullable: Boolean = !failOnError || children.exists(_.nullable)
override def dataType: DataType = VariantType
override def inputTypes: Seq[AbstractDataType] =
@@ -898,34 +868,46 @@ case class VariantInsert(input: Expression, path:
Expression, value: Expression)
foldablePath match {
case Some(parsed) =>
VariantExpressionEvalUtils.insertAtPath(
- inputVariant, parsed.javaSegments, parsed.pathStr, valValue,
value.dataType, prettyName)
+ inputVariant, parsed.javaSegments, parsed.pathStr, valValue,
value.dataType, prettyName,
+ failOnError)
case None =>
VariantExpressionEvalUtils.insertAtPath(
- inputVariant, p.asInstanceOf[UTF8String], valValue, value.dataType,
prettyName)
+ inputVariant, p.asInstanceOf[UTF8String], valValue, value.dataType,
prettyName,
+ failOnError)
}
}
override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = {
val cls = VariantExpressionEvalUtils.getClass.getName.stripSuffix("$")
- nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) => {
- val fromArg = ctx.addReferenceObj("from", value.dataType)
- foldablePath match {
- case Some(parsed) =>
- val parsedArg = ctx.addReferenceObj("insertPath", parsed)
- s"""
- |${ev.value} = $cls.insertAtPath(
- | $vVal, $parsedArg.javaSegments(), $parsedArg.pathStr(),
$valVal, $fromArg,
- | "$prettyName");
- """.stripMargin
- case None =>
- s"""
- |${ev.value} = $cls.insertAtPath($vVal, $pVal, $valVal, $fromArg,
"$prettyName");
- """.stripMargin
- }
- })
+ val fromArg = ctx.addReferenceObj("from", value.dataType)
+ def call(vVal: String, pVal: String, valVal: String): String =
foldablePath match {
+ case Some(parsed) =>
+ val parsedArg = ctx.addReferenceObj("insertPath", parsed)
+ s"""$cls.insertAtPath(
+ | $vVal, $parsedArg.javaSegments(), $parsedArg.pathStr(), $valVal,
$fromArg,
+ | "$prettyName", $failOnError)""".stripMargin
+ case None =>
+ s"""$cls.insertAtPath($vVal, $pVal, $valVal, $fromArg, "$prettyName",
$failOnError)"""
+ }
+ if (failOnError) {
+ nullSafeCodeGen(ctx, ev,
+ (vVal, pVal, valVal) => s"${ev.value} = ${call(vVal, pVal, valVal)};")
+ } else {
+ val resultType = CodeGenerator.javaType(VariantType)
+ val tmp = ctx.freshName("insertResult")
+ nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) =>
+ s"""
+ |$resultType $tmp = ${call(vVal, pVal, valVal)};
+ |if ($tmp == null) {
+ | ${ev.isNull} = true;
+ |} else {
+ | ${ev.value} = $tmp;
+ |}
+ """.stripMargin)
+ }
}
- override def prettyName: String = "variant_insert"
+ override def prettyName: String = if (failOnError) "variant_insert" else
"try_variant_insert"
override protected def withNewChildrenInternal(
newFirst: Expression, newSecond: Expression, newThird: Expression):
VariantInsert =
@@ -942,6 +924,93 @@ object VariantInsert {
}
}
+abstract class VariantInsertExpressionBuilderBase(failOnError: Boolean)
extends ExpressionBuilder {
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression = {
+ val numArgs = expressions.length
+ if (numArgs == 3) {
+ VariantInsert(expressions(0), expressions(1), expressions(2),
failOnError)
+ } else {
+ throw QueryCompilationErrors.wrongNumArgsError(funcName, Seq(3), numArgs)
+ }
+ }
+}
+
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = "_FUNC_(v, path, val) - Inserts a value into a variant at the given
JSONPath " +
+ "location. An object path adds a new field (error if it already exists);
an array path " +
+ "inserts at the index, shifting later elements right. Missing intermediate
keys are created. " +
+ "Throws an error if a path segment hits a value of an incompatible type.
Returns NULL if any " +
+ "argument is NULL.",
+ arguments = """
+ Arguments:
+ * v - A variant value to mutate.
+ * path - A string expression evaluating to a JSONPath identifying the
insertion target. A
+ valid path should start with `$` and is followed by one or more
segments like `[123]`,
+ `.name`, `['name']`, or `["name"]`. The root path `$` is not allowed.
+ * val - Any expression castable to variant.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_(parse_json('{"a": 1}'), '$.b', 2);
+ {"a":1,"b":2}
+ > SELECT _FUNC_(parse_json('{}'), '$.a.b', 1);
+ {"a":{"b":1}}
+ > SELECT _FUNC_(parse_json('["a","b","c"]'), '$[1]', 'z');
+ ["a","z","b","c"]
+ > SELECT _FUNC_(parse_json('["a","b"]'), '$[5]', 'z');
+ ["a","b",null,null,null,"z"]
+ > SELECT _FUNC_(parse_json('{}'), '$.a', parse_json('{"x":1}'));
+ {"a":{"x":1}}
+ > SELECT _FUNC_(parse_json('{}'), '$.a', parse_json('null'));
+ {"a":null}
+ > SELECT _FUNC_(parse_json('{}'), '$.a', null);
+ NULL
+ """,
+ since = "4.3.0",
+ group = "variant_funcs"
+)
+// scalastyle:on line.size.limit
+object VariantInsertExpressionBuilder extends
VariantInsertExpressionBuilderBase(true)
+
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = "_FUNC_(v, path, val) - Inserts a value into a variant at the given
JSONPath " +
+ "location. An object path adds a new field; an array path inserts at the
index, shifting " +
+ "later elements right. Missing intermediate keys are created. Returns NULL
if the field " +
+ "already exists or a path segment hits a value of an incompatible type, or
if any argument is " +
+ "NULL.",
+ arguments = """
+ Arguments:
+ * v - A variant value to mutate.
+ * path - A string expression evaluating to a JSONPath identifying the
insertion target. A
+ valid path should start with `$` and is followed by one or more
segments like `[123]`,
+ `.name`, `['name']`, or `["name"]`. The root path `$` is not allowed.
+ * val - Any expression castable to variant.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_(parse_json('{"a": 1}'), '$.b', 2);
+ {"a":1,"b":2}
+ > SELECT _FUNC_(parse_json('{}'), '$.a.b', 1);
+ {"a":{"b":1}}
+ > SELECT _FUNC_(parse_json('["a","b","c"]'), '$[1]', 'z');
+ ["a","z","b","c"]
+ > SELECT _FUNC_(parse_json('["a","b"]'), '$[5]', 'z');
+ ["a","b",null,null,null,"z"]
+ > SELECT _FUNC_(parse_json('{"a": 1}'), '$.a', 2);
+ NULL
+ > SELECT _FUNC_(parse_json('{"a": 1}'), '$.a.b', 2);
+ NULL
+ > SELECT _FUNC_(parse_json('{}'), '$.a', null);
+ NULL
+ """,
+ since = "4.3.0",
+ group = "variant_funcs"
+)
+// scalastyle:on line.size.limit
+object TryVariantInsertExpressionBuilder extends
VariantInsertExpressionBuilderBase(false)
+
case class VariantExplode(child: Expression) extends UnaryExpression with
Generator
with ExpectsInputTypes {
override def inputTypes: Seq[AbstractDataType] = Seq(VariantType)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala
index 6f1a410c5bb9..ec28ed514c19 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala
@@ -1329,11 +1329,48 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
test("variant_insert") {
def checkInsert(input: String, path: String, value: Expression, expected:
String): Unit = {
- val expr = VariantInsert(
- Literal(parseJson(input)), Literal.create(path, StringType), value)
+ Seq(true, false).foreach { failOnError =>
+ val expr = VariantInsert(
+ Literal(parseJson(input)), Literal.create(path, StringType), value,
failOnError)
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(Cast(expr, StringType)),
+ expected)
+ }
+ }
+
+ // A recoverable error: `variant_insert` throws, but `try_variant_insert`
returns NULL.
+ def checkInsertRecoverableError(
+ input: String,
+ path: String,
+ value: Expression,
+ condition: String,
+ parameters: Map[String, String]): Unit = {
+ checkErrorInExpression[SparkRuntimeException](
+ VariantInsert(Literal(parseJson(input)), Literal.create(path,
StringType), value),
+ condition,
+ parameters)
checkEvaluation(
- ResolveTimeZone.resolveTimeZones(Cast(expr, StringType)),
- expected)
+ VariantInsert(
+ Literal(parseJson(input)), Literal.create(path, StringType), value,
failOnError = false),
+ null)
+ }
+
+ // An unrecoverable error (a malformed path or a size overflow): both
functions throw, and only
+ // the `functionName` in the error message differs.
+ def checkInsertUnrecoverableError(
+ input: String,
+ path: String,
+ value: Expression,
+ condition: String,
+ parameters: String => Map[String, String]): Unit = {
+ Seq("variant_insert" -> true, "try_variant_insert" -> false).foreach {
+ case (name, failOnError) =>
+ checkErrorInExpression[SparkRuntimeException](
+ VariantInsert(
+ Literal(parseJson(input)), Literal.create(path, StringType),
value, failOnError),
+ condition,
+ parameters(name))
+ }
}
// Object inserts.
@@ -1385,55 +1422,48 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
checkInsert(
"{}", "$.a", Literal.create(Array(1, 2, 3), ArrayType(IntegerType)),
"""{"a":[1,2,3]}""")
- // NULL-intolerant: any NULL argument yields NULL.
- checkEvaluation(
- VariantInsert(Literal.create(null, VariantType), Literal("$.a"),
Literal(1)),
- null)
+ Seq(true, false).foreach { failOnError =>
+ checkEvaluation(
+ VariantInsert(Literal.create(null, VariantType), Literal("$.a"),
Literal(1), failOnError),
+ null)
+ }
checkInsert("""{"a": 1}""", null, Literal(1), null)
checkInsert("""{"a": 1}""", "$.b", Literal.create(null, VariantType), null)
checkInsert("""{"a": 1}""", "$.b", Literal.create(null, NullType), null)
- val dynamic = VariantInsert(
- Literal(parseJson("""{"a": 1}""")),
- BoundReference(0, StringType, nullable = true),
- Literal(2))
- checkEvaluation(
- ResolveTimeZone.resolveTimeZones(Cast(dynamic, StringType)),
- """{"a":1,"b":2}""",
- InternalRow(UTF8String.fromString("$.b")))
+ Seq(true, false).foreach { failOnError =>
+ val dynamic = VariantInsert(
+ Literal(parseJson("""{"a": 1}""")),
+ BoundReference(0, StringType, nullable = true),
+ Literal(2),
+ failOnError)
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(Cast(dynamic, StringType)),
+ """{"a":1,"b":2}""",
+ InternalRow(UTF8String.fromString("$.b")))
+ }
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a": 1}""")), Literal("$.a"),
Literal(2)),
+ // Recoverable errors: `variant_insert` throws; `try_variant_insert`
returns NULL. A duplicate
+ // key and every shape of path type mismatch are all recoverable.
+ checkInsertRecoverableError("""{"a": 1}""", "$.a", Literal(2),
"VARIANT_DUPLICATE_KEY",
Map("key" -> "a"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a": 1}""")), Literal("$.a.b"),
Literal(2)),
+ checkInsertRecoverableError("""{"a": 1}""", "$.a.b", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$.a.b", "failedAt" -> "$.a", "functionName" ->
"`variant_insert`"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a": 1}""")), Literal("$[0]"),
Literal(2)),
+ checkInsertRecoverableError("""{"a": 1}""", "$[0]", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$[0]", "failedAt" -> "$", "functionName" ->
"`variant_insert`"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a": [1, 2]}""")), Literal("$.a.b"),
Literal(2)),
+ checkInsertRecoverableError("""{"a": [1, 2]}""", "$.a.b", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$.a.b", "failedAt" -> "$.a", "functionName" ->
"`variant_insert`"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a": 5}""")), Literal("$.a[0]"),
Literal(2)),
+ checkInsertRecoverableError("""{"a": 5}""", "$.a[0]", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$.a[0]", "failedAt" -> "$.a", "functionName" ->
"`variant_insert`"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("5")), Literal("$.a"), Literal(2)),
+ checkInsertRecoverableError("5", "$.a", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$.a", "failedAt" -> "$", "functionName" ->
"`variant_insert`"))
-
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("""{"a.b": 5}""")),
Literal("""$['a.b'].c"""), Literal(2)),
+ checkInsertRecoverableError("""{"a.b": 5}""", """$['a.b'].c""", Literal(2),
"VARIANT_PATH_TYPE_MISMATCH",
Map("path" -> "$['a.b'].c", "failedAt" -> "$['a.b']", "functionName" ->
"`variant_insert`"))
@@ -1442,27 +1472,27 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
Literal.create(null, MapType(StringType, IntegerType)),
Literal.create(null, StructType(Seq(StructField("x", IntegerType))))
).foreach { v =>
- assert(
- VariantInsert(Literal(parseJson("{}")), Literal("$.a"),
v).checkInputDataTypes().isFailure)
+ Seq(true, false).foreach { failOnError =>
+ assert(
+ VariantInsert(Literal(parseJson("{}")), Literal("$.a"), v,
failOnError)
+ .checkInputDataTypes().isFailure)
+ }
}
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("{}")), Literal("$"), Literal(1)),
+ // Unrecoverable errors: both functions throw.
+ checkInsertUnrecoverableError("{}", "$", Literal(1),
"INVALID_VARIANT_PATH",
- Map("path" -> "$", "functionName" -> "`variant_insert`"))
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("{}")), Literal("abc"), Literal(1)),
+ name => Map("path" -> "$", "functionName" -> s"`$name`"))
+ checkInsertUnrecoverableError("{}", "abc", Literal(1),
"INVALID_VARIANT_PATH",
- Map("path" -> "abc", "functionName" -> "`variant_insert`"))
+ name => Map("path" -> "abc", "functionName" -> s"`$name`"))
val tooBig = "x".repeat(16 * 1024 * 1024)
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("{}")), Literal("$.a[2000000000]"),
Literal(1)),
+ checkInsertUnrecoverableError("{}", "$.a[2000000000]", Literal(1),
"VARIANT_SIZE_LIMIT",
- Map("sizeLimit" -> "16.0 MiB", "functionName" -> "`variant_insert`"))
- checkErrorInExpression[SparkRuntimeException](
- VariantInsert(Literal(parseJson("{}")), Literal("$.a"), Literal(tooBig)),
+ name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`"))
+ checkInsertUnrecoverableError("{}", "$.a", Literal(tooBig),
"VARIANT_SIZE_LIMIT",
- Map("sizeLimit" -> "16.0 MiB", "functionName" -> "`variant_insert`"))
+ name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`"))
}
}
diff --git
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
index 34faa4545b98..0222a40219ae 100644
---
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
+++
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
@@ -2767,6 +2767,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite
with Logging {
fn.variant_insert(fn.parse_json(fn.col("g")), "$.a", fn.lit(1))
}
+ functionTest("try_variant_insert") {
+ fn.try_variant_insert(fn.parse_json(fn.col("g")), "$.a", fn.lit(1))
+ }
+
functionTest("variant_get") {
fn.variant_get(fn.parse_json(fn.col("g")), "$", "int")
}
diff --git
a/sql/connect/common/src/test/resources/query-tests/explain-results/function_try_variant_insert.explain
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_try_variant_insert.explain
new file mode 100644
index 000000000000..cec39e83751e
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_try_variant_insert.explain
@@ -0,0 +1,2 @@
+Project
[try_variant_insert(static_invoke(VariantExpressionEvalUtils.parseJson(g#0,
false, true, true)), $.a, 1, false) AS try_variant_insert(parse_json(g), $.a,
1)#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]
diff --git
a/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_insert.explain
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_insert.explain
index 2b4cdcebf273..583443ce13fc 100644
---
a/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_insert.explain
+++
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_insert.explain
@@ -1,2 +1,2 @@
-Project
[variant_insert(static_invoke(VariantExpressionEvalUtils.parseJson(g#0, false,
true, true)), $.a, 1) AS variant_insert(parse_json(g), $.a, 1)#0]
+Project
[variant_insert(static_invoke(VariantExpressionEvalUtils.parseJson(g#0, false,
true, true)), $.a, 1, true) AS variant_insert(parse_json(g), $.a, 1)#0]
+- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]
diff --git
a/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.json
b/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.json
new file mode 100644
index 000000000000..f7c0e623a966
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.json
@@ -0,0 +1,125 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema":
"struct\u003cid:bigint,a:int,b:double,d:struct\u003cid:bigint,a:int,b:double\u003e,e:array\u003cint\u003e,f:map\u003cstring,struct\u003cid:bigint,a:int,b:double\u003e\u003e,g:string\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "try_variant_insert",
+ "arguments": [{
+ "unresolvedFunction": {
+ "functionName": "parse_json",
+ "arguments": [{
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "g"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass":
"org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "parse_json",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass":
"org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "literal": {
+ "string": "$.a"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "try_variant_insert",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass":
"org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "literal": {
+ "integer": 1
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "lit",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass":
"org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "try_variant_insert",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }]
+ }
+}
\ No newline at end of file
diff --git
a/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.proto.bin
b/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.proto.bin
new file mode 100644
index 000000000000..77bccb64f2c2
Binary files /dev/null and
b/sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.proto.bin
differ
diff --git a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
index 8e49cd70641e..ecdb01a2e034 100644
--- a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
+++ b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
@@ -555,9 +555,10 @@
| org.apache.spark.sql.catalyst.expressions.variant.ToVariantObject |
to_variant_object | SELECT to_variant_object(named_struct('a', 1, 'b', 2)) |
struct<to_variant_object(named_struct(a, 1, b, 2)):variant> |
|
org.apache.spark.sql.catalyst.expressions.variant.TryParseJsonExpressionBuilder
| try_parse_json | SELECT try_parse_json('{"a":1,"b":0.8}') |
struct<try_parse_json({"a":1,"b":0.8}):variant> |
|
org.apache.spark.sql.catalyst.expressions.variant.TryVariantGetExpressionBuilder
| try_variant_get | SELECT try_variant_get(parse_json('{"a": 1}'), '$.a',
'int') | struct<try_variant_get(parse_json({"a": 1}), $.a):int> |
+|
org.apache.spark.sql.catalyst.expressions.variant.TryVariantInsertExpressionBuilder
| try_variant_insert | SELECT try_variant_insert(parse_json('{"a": 1}'),
'$.b', 2) | struct<try_variant_insert(parse_json({"a": 1}), $.b, 2):variant> |
| org.apache.spark.sql.catalyst.expressions.variant.VariantDelete |
variant_delete | SELECT variant_delete(parse_json('{"a": 1, "b": 2, "c": 3,
"items": [1, 2, 3]}'), NULL, '$.a', '$.c') |
struct<variant_delete(parse_json({"a": 1, "b": 2, "c": 3, "items": [1, 2, 3]}),
NULL, $.a, $.c):variant> |
|
org.apache.spark.sql.catalyst.expressions.variant.VariantGetExpressionBuilder |
variant_get | SELECT variant_get(parse_json('{"a": 1}'), '$.a', 'int') |
struct<variant_get(parse_json({"a": 1}), $.a):int> |
-| org.apache.spark.sql.catalyst.expressions.variant.VariantInsert |
variant_insert | SELECT variant_insert(parse_json('{"a": 1}'), '$.b', 2) |
struct<variant_insert(parse_json({"a": 1}), $.b, 2):variant> |
+|
org.apache.spark.sql.catalyst.expressions.variant.VariantInsertExpressionBuilder
| variant_insert | SELECT variant_insert(parse_json('{"a": 1}'), '$.b', 2) |
struct<variant_insert(parse_json({"a": 1}), $.b, 2):variant> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathBoolean | xpath_boolean |
SELECT xpath_boolean('<a><b>1</b></a>','a/b') |
struct<xpath_boolean(<a><b>1</b></a>, a/b):boolean> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_double |
SELECT xpath_double('<a><b>1</b><b>2</b></a>', 'sum(a/b)') |
struct<xpath_double(<a><b>1</b><b>2</b></a>, sum(a/b)):double> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_number |
SELECT xpath_number('<a><b>1</b><b>2</b></a>', 'sum(a/b)') |
struct<xpath_number(<a><b>1</b><b>2</b></a>, sum(a/b)):double> |
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
index 0390507261f5..8d25633887f9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
@@ -384,6 +384,61 @@ class VariantSuite extends SharedSparkSession with
ExpressionEvalHelper {
}
}
+ test("try_variant_insert with literal arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b',
2))"),
+ rows("""{"a":1,"b":2}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.a',
2))"),
+ rows(null))
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'),
'$.a.b', 2))"),
+ rows(null))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$[0]',
2))"),
+ rows(null))
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 5}'),
'$.a[0]', 2))"),
+ rows(null))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b',
NULL))"),
+ rows(null))
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT try_variant_insert(parse_json('{}'), '$', 1)").collect()
+ },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "$", "functionName" ->
toSQLId("try_variant_insert")))
+ }
+
+ test("try_variant_insert with dynamic arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+ Seq("CODEGEN_ONLY", "NO_CODEGEN").foreach { codegenMode =>
+ withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> codegenMode) {
+ val df = Seq(
+ ("""{"a": 1}""", "$.b"),
+ ("""{"a": 1}""", "$.a"),
+ ("""{"a": 1}""", "$.a.b"),
+ (null, "$.b")
+ ).toDF("json", "path")
+ val v = parse_json(col("json"))
+ val out = df.select(to_json(try_variant_insert(v, col("path"),
lit(2))).alias("r"))
+ checkAnswer(out, rows("""{"a":1,"b":2}""", null, null, null))
+
+ val out2 = df.select(to_json(try_variant_insert(v, "$.b",
lit(2))).alias("r"))
+ checkAnswer(
+ out2,
+ rows("""{"a":1,"b":2}""", """{"a":1,"b":2}""", """{"a":1,"b":2}""",
null))
+ }
+ }
+ }
+
test("round trip tests") {
withSQLConf(SQLConf.VARIANT_INFER_SHREDDING_SCHEMA.key -> "false") {
val rand = new Random(42)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
