From f89f7df2dd283599db4158aa9ad8a4f70622d860 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bojana=20Ze=C4=8Devi=C4=87?= Date: Wed, 8 Jul 2026 13:07:50 +0000 Subject: [PATCH 1/4] init --- .../reference/pyspark.sql/functions.rst | 1 + .../pyspark/sql/connect/functions/builtin.py | 10 + python/pyspark/sql/functions/__init__.py | 1 + python/pyspark/sql/functions/builtin.py | 57 +++++- python/pyspark/sql/tests/test_functions.py | 6 + .../org/apache/spark/sql/functions.scala | 46 ++++- .../catalyst/analysis/FunctionRegistry.scala | 3 +- .../variant/VariantExpressionEvalUtils.scala | 17 +- .../variant/variantExpressions.scala | 176 ++++++++++++------ .../variant/VariantExpressionSuite.scala | 138 +++++++++----- .../spark/sql/PlanGenerationTestSuite.scala | 4 + .../function_try_variant_insert.explain | 2 + .../function_variant_insert.explain | 2 +- .../queries/function_try_variant_insert.json | 125 +++++++++++++ .../function_try_variant_insert.proto.bin | Bin 0 -> 1151 bytes .../sql-functions/sql-expression-schema.md | 3 +- .../org/apache/spark/sql/VariantSuite.scala | 59 ++++++ 17 files changed, 534 insertions(+), 116 deletions(-) create mode 100644 sql/connect/common/src/test/resources/query-tests/explain-results/function_try_variant_insert.explain create mode 100644 sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.json create mode 100644 sql/connect/common/src/test/resources/query-tests/queries/function_try_variant_insert.proto.bin diff --git a/python/docs/source/reference/pyspark.sql/functions.rst b/python/docs/source/reference/pyspark.sql/functions.rst index 70e5811180d0c..9ae6cd12a13c3 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 8e74aeef2828f..8a33d3b2b38e3 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 ee9d06fcaa2d8..052ca8a517370 100644 --- a/python/pyspark/sql/functions/__init__.py +++ b/python/pyspark/sql/functions/__init__.py @@ -483,6 +483,7 @@ "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 1e348cfd42874..ff85000545259 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 an + incompatible value. Returns NULL if any argument is NULL. .. versionadded:: 4.3.0 @@ -21786,6 +21787,60 @@ 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 an incompatible + value, 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, "$.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 2ade286763dea..0a4b4f5e6a502 100644 --- a/python/pyspark/sql/tests/test_functions.py +++ b/python/pyspark/sql/tests/test_functions.py @@ -3534,6 +3534,12 @@ def check(resultDf, expected): 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}'], + ) + # Duplicate key is caught -> NULL (df.path is an existing key in each row). + check(df.select(F.to_json(F.try_variant_insert(v, df.path, F.lit(9)))), [None, None]) check(df.select(F.schema_of_variant(v)), ["OBJECT", "OBJECT"]) check(df.select(F.schema_of_variant_agg(v)), ["OBJECT"]) 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 c8b269a63de5e..8494c8d2b5378 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 + * an incompatible value. 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 + * an incompatible value. 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 an incompatible + * value, 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 an incompatible + * value, 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 193e939889dbf..f1b788eb3e81e 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 20a55bccc7791..ce5fe47c8dac9 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,11 +152,14 @@ object VariantExpressionEvalUtils { val out = VariantBuilder.insertAtPath(v, javaSegments, valVariant) new VariantVal(out.getValue, out.getMetadata) } catch { + case _: VariantPathTypeMismatchException if !failOnError => null case e: VariantPathTypeMismatchException => throw QueryExecutionErrors.variantPathTypeMismatch( path, renderVariantPath(javaSegments.take(e.depth)), functionName) case _: VariantSizeLimitException => throw QueryExecutionErrors.variantSizeLimitError(VariantUtil.SIZE_LIMIT, functionName) + case e: SparkRuntimeException if !failOnError && e.getCondition == "VARIANT_DUPLICATE_KEY" => + null } } @@ -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 0f8995ce829fe..9aceae76b902d 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 { @@ -854,6 +823,10 @@ case class VariantInsert(input: Expression, path: Expression, value: Expression) override def nullIntolerant: Boolean = true + // `try_variant_insert` (failOnError = false) can return NULL for non-NULL inputs when it catches + // an error, so it is always nullable. + override def nullable: Boolean = !failOnError || children.exists(_.nullable) + override def dataType: DataType = VariantType override def inputTypes: Seq[AbstractDataType] = Seq(VariantType, StringTypeWithCollation(supportsTrimCollation = true), AnyDataType) @@ -898,34 +871,48 @@ 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) + // The `insertAtPath(...)` call for the evaluated child value names. + 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) { + // `variant_insert` never returns null for non-null inputs. + nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) => s"${ev.value} = ${call(vVal, pVal, valVal)};") + } else { + // `try_variant_insert` may return null (a caught error). It is always nullable, so `ev.isNull` + // is a mutable variable that the guard below can set. + 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 +929,85 @@ 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 an incompatible value. 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 an incompatible value, 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": 1}'), '$.a', 2); + NULL + > SELECT _FUNC_(parse_json('{"a": 1}'), '$.a.b', 2); + 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 6f1a410c5bb91..142b681a09282 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 @@ -1328,12 +1328,51 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { } test("variant_insert") { + // Every success case is asserted for both `variant_insert` and `try_variant_insert`: the two + // agree whenever the insertion is valid. 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 catchable error: `variant_insert` throws, but `try_variant_insert` returns NULL. + def checkInsertCaughtError( + 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) + } + + // A non-catchable error (a malformed path or a size overflow): both functions throw, and only + // the `functionName` in the error message differs. + def checkInsertUncaughtError( + 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,84 +1424,81 @@ 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) + // NULL-intolerant: any NULL argument yields NULL (for both functions). + 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"))) + // Dynamic (non-foldable) path, for both functions. + 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)), + // Catchable errors: `variant_insert` throws; `try_variant_insert` returns NULL. A duplicate + // key and every shape of path type mismatch are all caught. + checkInsertCaughtError("""{"a": 1}""", "$.a", Literal(2), "VARIANT_DUPLICATE_KEY", Map("key" -> "a")) - - checkErrorInExpression[SparkRuntimeException]( - VariantInsert(Literal(parseJson("""{"a": 1}""")), Literal("$.a.b"), Literal(2)), + checkInsertCaughtError("""{"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)), + checkInsertCaughtError("""{"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)), + checkInsertCaughtError("""{"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)), + checkInsertCaughtError("""{"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)), + checkInsertCaughtError("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)), + checkInsertCaughtError("""{"a.b": 5}""", """$['a.b'].c""", Literal(2), "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$['a.b'].c", "failedAt" -> "$['a.b']", "functionName" -> "`variant_insert`")) - // Structs and maps are rejected at analysis. + // Structs and maps are rejected at analysis for both functions. Seq( 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)), + // Non-catchable errors: both functions throw. A malformed path is rejected during parsing + // (before the insert), and a size overflow is raised whether it comes from a huge structure + // or a huge value. + checkInsertUncaughtError("{}", "$", 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`")) + checkInsertUncaughtError("{}", "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)), + checkInsertUncaughtError("{}", "$.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`")) + checkInsertUncaughtError("{}", "$.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 34faa4545b98f..0222a40219aea 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 0000000000000..cec39e83751ea --- /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 , [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 2b4cdcebf273a..583443ce13fc7 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 , [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 0000000000000..f7c0e623a9668 --- /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 0000000000000000000000000000000000000000..77bccb64f2c2b80d0ea96cc27180a03a3d67ca37 GIT binary patch literal 1151 zcmdT?KTE?v9PO{cKipzDw}1%JK}rbHF@y|paBvZHaw)$i*LXF(#Jh{oC4-aTmc6qe zC;!tUEDK0l0@4b8P-XqNx*q`g90@7?XPH%u-gVM5yw9jL&&l4`R zgI<01y*MlSDRW|P4kB=vhf?C$H@v_Zcq7bx!{H)v=1~JDspSncus_!9$ts~*j(e9V zISSq5LNTf7m1VyJAvsFLdP9{7`U2D3w>~8&no<6kOsNf06BhUv3_ZPJMBlz;1Q`5l8&}zF_!5ffQFdc-43nrLR>JpPY VoAvL(U$&ESy*%IbcK-Lze*p^6ndSfh literal 0 HcmV?d00001 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 8e49cd70641e9..ecdb01a2e0345 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 | | org.apache.spark.sql.catalyst.expressions.variant.TryParseJsonExpressionBuilder | try_parse_json | SELECT try_parse_json('{"a":1,"b":0.8}') | struct | | org.apache.spark.sql.catalyst.expressions.variant.TryVariantGetExpressionBuilder | try_variant_get | SELECT try_variant_get(parse_json('{"a": 1}'), '$.a', 'int') | struct | +| org.apache.spark.sql.catalyst.expressions.variant.TryVariantInsertExpressionBuilder | try_variant_insert | SELECT try_variant_insert(parse_json('{"a": 1}'), '$.b', 2) | struct | | 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 | | org.apache.spark.sql.catalyst.expressions.variant.VariantGetExpressionBuilder | variant_get | SELECT variant_get(parse_json('{"a": 1}'), '$.a', 'int') | struct | -| org.apache.spark.sql.catalyst.expressions.variant.VariantInsert | variant_insert | SELECT variant_insert(parse_json('{"a": 1}'), '$.b', 2) | struct | +| org.apache.spark.sql.catalyst.expressions.variant.VariantInsertExpressionBuilder | variant_insert | SELECT variant_insert(parse_json('{"a": 1}'), '$.b', 2) | struct | | org.apache.spark.sql.catalyst.expressions.xml.XPathBoolean | xpath_boolean | SELECT xpath_boolean('1','a/b') | struct1, a/b):boolean> | | org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_double | SELECT xpath_double('12', 'sum(a/b)') | struct12, sum(a/b)):double> | | org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_number | SELECT xpath_number('12', 'sum(a/b)') | struct12, 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 0390507261f55..0d24769c7c1c4 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,65 @@ class VariantSuite extends SharedSparkSession with ExpressionEvalHelper { } } + test("try_variant_insert with literal arguments") { + def rows(results: Any*): Seq[Row] = results.map(Row(_)) + + // Succeeds like variant_insert. + checkAnswer( + sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b', 2))"), + rows("""{"a":1,"b":2}""")) + // A duplicate key or any shape of path type conflict is caught -> NULL. + 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)) + // An array index into an object, and an object descent into a scalar, are both caught -> 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)) + // NULL-intolerant. + checkAnswer( + sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b', NULL))"), + rows(null)) + + // A malformed path is NOT caught -> still throws. + 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")) + // Dynamic path: success, duplicate key (NULL), type conflict (NULL), NULL input (NULL). + 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)) + + // String-path overload of the DataFrame API. + 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) From feb7418c8e84e0e955155f98f0e456bab9b66c76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bojana=20Ze=C4=8Devi=C4=87?= Date: Wed, 8 Jul 2026 15:41:09 +0000 Subject: [PATCH 2/4] improve --- python/pyspark/sql/functions/builtin.py | 14 +++++++++---- python/pyspark/sql/tests/test_functions.py | 2 +- .../org/apache/spark/sql/functions.scala | 12 +++++------ .../variant/variantExpressions.scala | 21 ++++++++++--------- .../variant/VariantExpressionSuite.scala | 10 ++------- .../org/apache/spark/sql/VariantSuite.scala | 10 +++------ 6 files changed, 33 insertions(+), 36 deletions(-) diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py index ff85000545259..2103822179d6c 100644 --- a/python/pyspark/sql/functions/builtin.py +++ b/python/pyspark/sql/functions/builtin.py @@ -21731,8 +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. Throws an error if a path segment hits an - incompatible value. 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 @@ -21794,8 +21794,8 @@ def try_variant_insert( """ 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 an incompatible - value, or if any argument is NULL. + 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 @@ -21823,6 +21823,12 @@ def try_variant_insert( >>> 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() diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py index 0a4b4f5e6a502..3ac206d71f92a 100644 --- a/python/pyspark/sql/tests/test_functions.py +++ b/python/pyspark/sql/tests/test_functions.py @@ -3538,8 +3538,8 @@ def check(resultDf, expected): df.select(F.to_json(F.try_variant_insert(v, "$.z", F.lit(9)))), ['{"a":1,"z":9}', '{"b":2,"z":9}'], ) - # Duplicate key is caught -> NULL (df.path is an existing key in each row). 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", "OBJECT"]) check(df.select(F.schema_of_variant_agg(v)), ["OBJECT"]) 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 8494c8d2b5378..967554a51cee3 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 @@ -9825,7 +9825,7 @@ 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. Throws an error if a path segment hits - * an incompatible value. Returns NULL if any argument is NULL. + * a value of an incompatible type. Returns NULL if any argument is NULL. * * @param v * a variant column. @@ -9845,7 +9845,7 @@ 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. Throws an error if a path segment hits - * an incompatible value. Returns NULL if any argument is NULL. + * a value of an incompatible type. Returns NULL if any argument is NULL. * * @param v * a variant column. @@ -9864,8 +9864,8 @@ object functions { /** * 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 an incompatible - * value, or if any argument is NULL. + * 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. @@ -9884,8 +9884,8 @@ object functions { /** * 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 an incompatible - * value, or if any argument is NULL. + * 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. 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 9aceae76b902d..3c3064db6f5bb 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 @@ -822,9 +822,6 @@ case class VariantInsert( override def third: Expression = value override def nullIntolerant: Boolean = true - - // `try_variant_insert` (failOnError = false) can return NULL for non-NULL inputs when it catches - // an error, so it is always nullable. override def nullable: Boolean = !failOnError || children.exists(_.nullable) override def dataType: DataType = VariantType @@ -882,7 +879,6 @@ case class VariantInsert( override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = { val cls = VariantExpressionEvalUtils.getClass.getName.stripSuffix("$") val fromArg = ctx.addReferenceObj("from", value.dataType) - // The `insertAtPath(...)` call for the evaluated child value names. def call(vVal: String, pVal: String, valVal: String): String = foldablePath match { case Some(parsed) => val parsedArg = ctx.addReferenceObj("insertPath", parsed) @@ -893,11 +889,8 @@ case class VariantInsert( s"""$cls.insertAtPath($vVal, $pVal, $valVal, $fromArg, "$prettyName", $failOnError)""" } if (failOnError) { - // `variant_insert` never returns null for non-null inputs. nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) => s"${ev.value} = ${call(vVal, pVal, valVal)};") } else { - // `try_variant_insert` may return null (a caught error). It is always nullable, so `ev.isNull` - // is a mutable variable that the guard below can set. val resultType = CodeGenerator.javaType(VariantType) val tmp = ctx.freshName("insertResult") nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) => @@ -945,8 +938,8 @@ abstract class VariantInsertExpressionBuilderBase(failOnError: Boolean) extends 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 an incompatible value. Returns NULL if any argument " + - "is NULL.", + "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. @@ -983,7 +976,7 @@ object VariantInsertExpressionBuilder extends VariantInsertExpressionBuilderBase 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 an incompatible value, or if any argument is " + + "already exists or a path segment hits a value of an incompatible type, or if any argument is " + "NULL.", arguments = """ Arguments: @@ -997,10 +990,18 @@ object VariantInsertExpressionBuilder extends VariantInsertExpressionBuilderBase 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" 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 142b681a09282..1a241c862ca96 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 @@ -1328,8 +1328,6 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { } test("variant_insert") { - // Every success case is asserted for both `variant_insert` and `try_variant_insert`: the two - // agree whenever the insertion is valid. def checkInsert(input: String, path: String, value: Expression, expected: String): Unit = { Seq(true, false).foreach { failOnError => val expr = VariantInsert( @@ -1424,7 +1422,6 @@ 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 (for both functions). Seq(true, false).foreach { failOnError => checkEvaluation( VariantInsert(Literal.create(null, VariantType), Literal("$.a"), Literal(1), failOnError), @@ -1434,7 +1431,6 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { checkInsert("""{"a": 1}""", "$.b", Literal.create(null, VariantType), null) checkInsert("""{"a": 1}""", "$.b", Literal.create(null, NullType), null) - // Dynamic (non-foldable) path, for both functions. Seq(true, false).foreach { failOnError => val dynamic = VariantInsert( Literal(parseJson("""{"a": 1}""")), @@ -1471,7 +1467,7 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$['a.b'].c", "failedAt" -> "$['a.b']", "functionName" -> "`variant_insert`")) - // Structs and maps are rejected at analysis for both functions. + // Structs and maps are rejected at analysis. Seq( Literal.create(null, MapType(StringType, IntegerType)), Literal.create(null, StructType(Seq(StructField("x", IntegerType)))) @@ -1483,9 +1479,7 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { } } - // Non-catchable errors: both functions throw. A malformed path is rejected during parsing - // (before the insert), and a size overflow is raised whether it comes from a huge structure - // or a huge value. + // Non-catchable errors: both functions throw. checkInsertUncaughtError("{}", "$", Literal(1), "INVALID_VARIANT_PATH", name => Map("path" -> "$", "functionName" -> s"`$name`")) 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 0d24769c7c1c4..8d25633887f95 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 @@ -387,30 +387,28 @@ class VariantSuite extends SharedSparkSession with ExpressionEvalHelper { test("try_variant_insert with literal arguments") { def rows(results: Any*): Seq[Row] = results.map(Row(_)) - // Succeeds like variant_insert. checkAnswer( sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b', 2))"), rows("""{"a":1,"b":2}""")) - // A duplicate key or any shape of path type conflict is caught -> NULL. + 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)) - // An array index into an object, and an object descent into a scalar, are both caught -> 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)) - // NULL-intolerant. + checkAnswer( sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b', NULL))"), rows(null)) - // A malformed path is NOT caught -> still throws. checkError( exception = intercept[SparkRuntimeException] { sql("SELECT try_variant_insert(parse_json('{}'), '$', 1)").collect() @@ -430,11 +428,9 @@ class VariantSuite extends SharedSparkSession with ExpressionEvalHelper { (null, "$.b") ).toDF("json", "path") val v = parse_json(col("json")) - // Dynamic path: success, duplicate key (NULL), type conflict (NULL), NULL input (NULL). 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)) - // String-path overload of the DataFrame API. val out2 = df.select(to_json(try_variant_insert(v, "$.b", lit(2))).alias("r")) checkAnswer( out2, From af1cd9d3e021b074f190ffa6c4081902a4bbb20f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bojana=20Ze=C4=8Devi=C4=87?= Date: Thu, 9 Jul 2026 10:10:08 +0000 Subject: [PATCH 3/4] address comments --- .../variant/VariantExpressionEvalUtils.scala | 4 +-- .../variant/VariantExpressionSuite.scala | 36 +++++++++---------- 2 files changed, 20 insertions(+), 20 deletions(-) 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 ce5fe47c8dac9..bf5d22ee7482a 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 @@ -153,13 +153,13 @@ object VariantExpressionEvalUtils { 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) case _: VariantSizeLimitException => throw QueryExecutionErrors.variantSizeLimitError(VariantUtil.SIZE_LIMIT, functionName) - case e: SparkRuntimeException if !failOnError && e.getCondition == "VARIANT_DUPLICATE_KEY" => - null } } 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 1a241c862ca96..ec28ed514c198 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 @@ -1338,8 +1338,8 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { } } - // A catchable error: `variant_insert` throws, but `try_variant_insert` returns NULL. - def checkInsertCaughtError( + // A recoverable error: `variant_insert` throws, but `try_variant_insert` returns NULL. + def checkInsertRecoverableError( input: String, path: String, value: Expression, @@ -1355,9 +1355,9 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { null) } - // A non-catchable error (a malformed path or a size overflow): both functions throw, and only + // An unrecoverable error (a malformed path or a size overflow): both functions throw, and only // the `functionName` in the error message differs. - def checkInsertUncaughtError( + def checkInsertUnrecoverableError( input: String, path: String, value: Expression, @@ -1443,27 +1443,27 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { InternalRow(UTF8String.fromString("$.b"))) } - // Catchable errors: `variant_insert` throws; `try_variant_insert` returns NULL. A duplicate - // key and every shape of path type mismatch are all caught. - checkInsertCaughtError("""{"a": 1}""", "$.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")) - checkInsertCaughtError("""{"a": 1}""", "$.a.b", Literal(2), + checkInsertRecoverableError("""{"a": 1}""", "$.a.b", Literal(2), "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$.a.b", "failedAt" -> "$.a", "functionName" -> "`variant_insert`")) - checkInsertCaughtError("""{"a": 1}""", "$[0]", Literal(2), + checkInsertRecoverableError("""{"a": 1}""", "$[0]", Literal(2), "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$[0]", "failedAt" -> "$", "functionName" -> "`variant_insert`")) - checkInsertCaughtError("""{"a": [1, 2]}""", "$.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`")) - checkInsertCaughtError("""{"a": 5}""", "$.a[0]", Literal(2), + checkInsertRecoverableError("""{"a": 5}""", "$.a[0]", Literal(2), "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$.a[0]", "failedAt" -> "$.a", "functionName" -> "`variant_insert`")) - checkInsertCaughtError("5", "$.a", Literal(2), + checkInsertRecoverableError("5", "$.a", Literal(2), "VARIANT_PATH_TYPE_MISMATCH", Map("path" -> "$.a", "failedAt" -> "$", "functionName" -> "`variant_insert`")) - checkInsertCaughtError("""{"a.b": 5}""", """$['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`")) @@ -1479,19 +1479,19 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { } } - // Non-catchable errors: both functions throw. - checkInsertUncaughtError("{}", "$", Literal(1), + // Unrecoverable errors: both functions throw. + checkInsertUnrecoverableError("{}", "$", Literal(1), "INVALID_VARIANT_PATH", name => Map("path" -> "$", "functionName" -> s"`$name`")) - checkInsertUncaughtError("{}", "abc", Literal(1), + checkInsertUnrecoverableError("{}", "abc", Literal(1), "INVALID_VARIANT_PATH", name => Map("path" -> "abc", "functionName" -> s"`$name`")) val tooBig = "x".repeat(16 * 1024 * 1024) - checkInsertUncaughtError("{}", "$.a[2000000000]", Literal(1), + checkInsertUnrecoverableError("{}", "$.a[2000000000]", Literal(1), "VARIANT_SIZE_LIMIT", name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`")) - checkInsertUncaughtError("{}", "$.a", Literal(tooBig), + checkInsertUnrecoverableError("{}", "$.a", Literal(tooBig), "VARIANT_SIZE_LIMIT", name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`")) } From bd403be447f5a98b3a3004452cd5bf5dee9f8679 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bojana=20Ze=C4=8Devi=C4=87?= Date: Thu, 9 Jul 2026 10:19:44 +0000 Subject: [PATCH 4/4] lint and format --- .../scala/org/apache/spark/sql/functions.scala | 16 ++++++++-------- .../expressions/variant/variantExpressions.scala | 6 ++++-- 2 files changed, 12 insertions(+), 10 deletions(-) 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 967554a51cee3..0be5b0b30c01b 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 @@ -9862,10 +9862,10 @@ object functions { 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. + * 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. @@ -9882,10 +9882,10 @@ object functions { 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. + * 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. 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 3c3064db6f5bb..3e404cc9c190a 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 @@ -872,7 +872,8 @@ case class VariantInsert( failOnError) case None => VariantExpressionEvalUtils.insertAtPath( - inputVariant, p.asInstanceOf[UTF8String], valValue, value.dataType, prettyName, failOnError) + inputVariant, p.asInstanceOf[UTF8String], valValue, value.dataType, prettyName, + failOnError) } } @@ -889,7 +890,8 @@ case class VariantInsert( s"""$cls.insertAtPath($vVal, $pVal, $valVal, $fromArg, "$prettyName", $failOnError)""" } if (failOnError) { - nullSafeCodeGen(ctx, ev, (vVal, pVal, valVal) => s"${ev.value} = ${call(vVal, pVal, valVal)};") + nullSafeCodeGen(ctx, ev, + (vVal, pVal, valVal) => s"${ev.value} = ${call(vVal, pVal, valVal)};") } else { val resultType = CodeGenerator.javaType(VariantType) val tmp = ctx.freshName("insertResult")