This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 38e2143defaa [SPARK-57417][SQL] Add `variant_delete` expression
38e2143defaa is described below
commit 38e2143defaa986472d37f7b28867b89290b5190
Author: bojana-db <[email protected]>
AuthorDate: Wed Jun 17 15:00:01 2026 -0700
[SPARK-57417][SQL] Add `variant_delete` expression
### What changes were proposed in this pull request?
Adds the SQL function `variant_delete(v, path1[, path2, …])`, which removes
fields or array elements from a Variant value at one or more JSONPath locations.
Details:
Multiple paths are applied left to right;
NULL paths are skipped; a NULL variant input returns NULL;
paths that don't match (missing key, out-of-range index, type mismatch) are
silent no-ops
### Why are the changes needed?
There is no way to delete a field from variant, the only way is to convert
it to other datatype (ex. map).
### Does this PR introduce _any_ user-facing change?
Yes, new SQL function.
### How was this patch tested?
Unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 4.8
Closes #56471 from bojana-db/variant-delete.
Authored-by: bojana-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 77d302a1d5aebf81962c72be314663df62fa8717)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 7 +
.../apache/spark/types/variant/VariantBuilder.java | 91 +++++++++++++
.../source/reference/pyspark.sql/functions.rst | 1 +
python/pyspark/sql/connect/functions/builtin.py | 13 ++
python/pyspark/sql/functions/__init__.py | 1 +
python/pyspark/sql/functions/builtin.py | 62 +++++++++
python/pyspark/sql/tests/test_functions.py | 6 +
.../scala/org/apache/spark/sql/functions.scala | 38 ++++++
.../sql/catalyst/analysis/FunctionRegistry.scala | 1 +
.../variant/VariantExpressionEvalUtils.scala | 30 +++++
.../expressions/variant/variantExpressions.scala | 142 +++++++++++++++++++++
.../spark/sql/errors/QueryExecutionErrors.scala | 6 +
.../variant/VariantExpressionSuite.scala | 111 ++++++++++++++++
.../apache/spark/sql/PlanGenerationTestSuite.scala | 4 +
.../function_variant_delete.explain | 2 +
.../queries/function_variant_delete.json | 125 ++++++++++++++++++
.../queries/function_variant_delete.proto.bin | Bin 0 -> 1153 bytes
.../sql-functions/sql-expression-schema.md | 1 +
.../scala/org/apache/spark/sql/VariantSuite.scala | 99 ++++++++++++++
19 files changed, 740 insertions(+)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 411cddbd16f7..a1d80b1ea9d1 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -5162,6 +5162,13 @@
],
"sqlState" : "22023"
},
+ "INVALID_VARIANT_PATH" : {
+ "message" : [
+ "The path '<path>' is not a valid <functionName> target.",
+ "A valid path should start with `$` and is followed by segments like
`[123]`, `.name`, `['name']`, or `[\"name\"]`."
+ ],
+ "sqlState" : "22023"
+ },
"INVALID_VARIANT_SHREDDING_SCHEMA" : {
"message" : [
"The schema `<schema>` is not a valid variant shredding schema."
diff --git
a/common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java
b/common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java
index aaf6f72bd536..dc0a114706c7 100644
---
a/common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java
+++
b/common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java
@@ -97,6 +97,40 @@ public class VariantBuilder {
return builder.result();
}
+ // A segment in a JSONPath used by variant manipulation functions.
+ public abstract static class PathSegment {}
+
+ // Object field access (e.g. `.a` or `['a']`).
+ public static final class ObjectKeySegment extends PathSegment {
+ public final String key;
+
+ public ObjectKeySegment(String key) {
+ this.key = key;
+ }
+ }
+
+ // Array index access (e.g. `[0]`).
+ public static final class ArrayIndexSegment extends PathSegment {
+ public final int index;
+
+ public ArrayIndexSegment(int index) {
+ this.index = index;
+ }
+ }
+
+ // Return a new variant with the field or array element at `segments`
removed. If the path does
+ // not match (missing key, out-of-range index, or incompatible container
type), a semantically
+ // equivalent variant is returned. The result is always rebuilt with fresh
metadata, so the
+ // binary representation may differ even when nothing is deleted. `segments`
must be non-empty.
+ public static Variant deleteAtPath(Variant v, PathSegment[] segments) {
+ if (segments.length == 0) {
+ throw new IllegalArgumentException("Segments must be non-empty");
+ }
+ VariantBuilder builder = new VariantBuilder(false);
+ builder.appendWithDeletionImpl(v.value, v.metadata, v.pos, segments, 0);
+ return builder.result();
+ }
+
// Build the variant metadata from `dictionaryKeys` and return the variant
result.
public Variant result() {
int numKeys = dictionaryKeys.size();
@@ -459,6 +493,63 @@ public class VariantBuilder {
}
}
+ private void appendWithDeletionImpl(
+ byte[] value, byte[] metadata, int pos, PathSegment[] segments, int
depth) {
+ checkIndex(pos, value.length);
+ PathSegment seg = segments[depth];
+ boolean isLast = depth == segments.length - 1;
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ if (seg instanceof ObjectKeySegment && basicType == OBJECT) {
+ String key = ((ObjectKeySegment) seg).key;
+ handleObject(value, pos, (size, idSize, offsetSize, idStart,
offsetStart, dataStart) -> {
+ ArrayList<FieldEntry> fields = new ArrayList<>(size);
+ int start = writePos;
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ String fieldKey = getMetadataKey(metadata, id);
+ boolean isTarget = fieldKey.equals(key);
+ if (!(isTarget && isLast)) {
+ int newId = addKey(fieldKey);
+ fields.add(new FieldEntry(fieldKey, newId, writePos - start));
+ if (isTarget) {
+ appendWithDeletionImpl(value, metadata, elementPos, segments,
depth + 1);
+ } else {
+ appendVariantImpl(value, metadata, elementPos);
+ }
+ }
+ }
+ finishWritingObject(start, fields);
+ return null;
+ });
+ } else if (seg instanceof ArrayIndexSegment && basicType == ARRAY) {
+ int index = ((ArrayIndexSegment) seg).index;
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ ArrayList<Integer> offsets = new ArrayList<>(size);
+ int start = writePos;
+ for (int i = 0; i < size; ++i) {
+ boolean isTarget = i == index;
+ if (!(isTarget && isLast)) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ offsets.add(writePos - start);
+ if (isTarget) {
+ appendWithDeletionImpl(value, metadata, elementPos, segments,
depth + 1);
+ } else {
+ appendVariantImpl(value, metadata, elementPos);
+ }
+ }
+ }
+ finishWritingArray(start, offsets);
+ return null;
+ });
+ } else {
+ // Container type does not match the segment kind; append unchanged.
+ appendVariantImpl(value, metadata, pos);
+ }
+ }
+
// Append the variant value without rewriting or creating any metadata. This
is used when
// building an object during shredding, where there is a fixed pre-existing
metadata that
// all shredded values will refer to.
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst
b/python/docs/source/reference/pyspark.sql/functions.rst
index 4b9084547b9a..e41066256bd4 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -600,6 +600,7 @@ VARIANT Functions
schema_of_variant
schema_of_variant_agg
try_variant_get
+ variant_delete
variant_get
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 11ec77c7d30c..828ac151cd8c 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -2206,6 +2206,19 @@ def is_valid_variant(v: "ColumnOrName") -> Column:
is_valid_variant.__doc__ = pysparkfuncs.is_valid_variant.__doc__
+def variant_delete(v: "ColumnOrName", *paths: Union[Column, str]) -> Column:
+ if len(paths) == 0:
+ raise PySparkValueError(
+ errorClass="CANNOT_BE_EMPTY",
+ messageParameters={"item": "paths"},
+ )
+ cols = [p if isinstance(p, Column) else lit(p) for p in paths]
+ return _invoke_function("variant_delete", _to_col(v), *cols)
+
+
+variant_delete.__doc__ = pysparkfuncs.variant_delete.__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 3801ba0e51e5..6d1f794474eb 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -478,6 +478,7 @@ __all__ = [ # noqa: F405
"schema_of_variant",
"schema_of_variant_agg",
"try_variant_get",
+ "variant_delete",
"variant_get",
"try_parse_json",
"to_variant_object",
diff --git a/python/pyspark/sql/functions/builtin.py
b/python/pyspark/sql/functions/builtin.py
index 59fc22cb43f1..f62dc1d93520 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -21513,6 +21513,68 @@ def is_valid_variant(v: "ColumnOrName") -> Column:
return _invoke_function("is_valid_variant", _to_java_column(v))
+@_try_remote_functions
+def variant_delete(v: "ColumnOrName", *paths: Union[Column, str]) -> Column:
+ """
+ Removes fields or array elements from a variant at the given JSONPath
locations.
+ Multiple paths are applied left to right. Returns NULL if `v` is NULL;
NULL paths are
+ skipped.
+
+ .. versionadded:: 5.0.0
+
+ Parameters
+ ----------
+ v : :class:`~pyspark.sql.Column` or str
+ a variant column or column name
+ paths : :class:`~pyspark.sql.Column` or str
+ one or more JSONPath deletion targets. 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.
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ a variant column with the specified paths removed
+
+ Examples
+ --------
+ >>> from pyspark.sql.functions import lit, parse_json, to_json,
variant_delete
+ >>> df = spark.createDataFrame([{
+ ... 'json': '''{ "a" : 1, "b" : 2, "c" : 3, "items" : [1, 2, 3] }''',
+ ... 'path': '$.a'
+ ... }])
+ >>> v = parse_json(df.json)
+ >>> df.select(to_json(variant_delete(v, lit(None), "$.a",
"$.c")).alias("r")).collect()
+ [Row(r='{"b":2,"items":[1,2,3]}')]
+ >>> df.select(to_json(variant_delete(v, "$.missing")).alias("r")).collect()
+ [Row(r='{"a":1,"b":2,"c":3,"items":[1,2,3]}')]
+ >>> df.select(to_json(variant_delete(v, df.path)).alias("r")).collect()
+ [Row(r='{"b":2,"c":3,"items":[1,2,3]}')]
+ >>> df.select(to_json(variant_delete(v, "$.items[0]",
"$.items[0]")).alias("r")).collect()
+ [Row(r='{"a":1,"b":2,"c":3,"items":[3]}')]
+ >>> df.select(variant_delete(lit(None), "$.a").alias("r")).collect()
+ [Row(r=None)]
+ """
+ from pyspark.sql.classic.column import _to_java_column, _to_seq
+
+ if len(paths) == 0:
+ raise PySparkValueError(
+ errorClass="CANNOT_BE_EMPTY",
+ messageParameters={"item": "paths"},
+ )
+ sc = _get_active_spark_context()
+
+ path_cols = [p if isinstance(p, Column) else lit(p) for p in paths]
+ return _invoke_function(
+ "variant_delete",
+ _to_java_column(v),
+ _to_java_column(path_cols[0]),
+ _to_seq(sc, path_cols[1:], _to_java_column),
+ )
+
+
@_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 4edb467d2c2b..8599d0dd46e1 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -3516,6 +3516,12 @@ class FunctionsTestsMixin:
check(df.select(F.is_variant_null(v)), [False, False])
check(df.select(F.is_valid_variant(v)), [True, True])
+ check(df.select(F.to_json(F.variant_delete(v, "$.a"))), ["{}",
'{"b":2}'])
+ check(df.select(F.to_json(F.variant_delete(v, df.path))), ["{}", "{}"])
+ check(
+ df.select(F.to_json(F.variant_delete(v, F.lit(None)))),
+ ['{"a":1}', '{"b":2}'],
+ )
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 fe91d0bd7b69..a2850f37a869 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
@@ -9688,6 +9688,44 @@ object functions {
*/
def is_valid_variant(v: Column): Column = Column.fn("is_valid_variant", v)
+ /**
+ * Removes fields or array elements from a variant at the given JSONPath
locations. Multiple
+ * paths are applied left to right. Returns NULL if `v` is NULL; NULL paths
are skipped.
+ *
+ * @param v
+ * a variant column.
+ * @param path
+ * the column containing the first JSONPath string. 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 paths
+ * additional JSONPath arguments, applied after `path` in order.
+ * @group variant_funcs
+ * @since 5.0.0
+ */
+ @scala.annotation.varargs
+ def variant_delete(v: Column, path: Column, paths: Column*): Column =
+ Column.fn("variant_delete", (v +: path +: paths): _*)
+
+ /**
+ * Removes fields or array elements from a variant at the given JSONPath
locations. Multiple
+ * paths are applied left to right. Returns NULL if `v` is NULL; NULL paths
are skipped.
+ *
+ * @param v
+ * a variant column.
+ * @param path
+ * the first JSONPath identifying a deletion 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 paths
+ * additional JSONPath strings, applied after `path` in order.
+ * @group variant_funcs
+ * @since 5.0.0
+ */
+ @scala.annotation.varargs
+ def variant_delete(v: Column, path: String, paths: String*): Column =
+ Column.fn("variant_delete", (v +: lit(path) +: paths.map(lit)): _*)
+
/**
* 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 1f216ad6a033..dea90247cc1b 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
@@ -982,6 +982,7 @@ object FunctionRegistry {
expression[SchemaOfVariantAgg]("schema_of_variant_agg"),
expression[ToVariantObject]("to_variant_object"),
expression[IsValidVariant]("is_valid_variant"),
+ expression[VariantDelete]("variant_delete"),
// 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 b32064ce78c1..47afbcd78837 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
@@ -75,6 +75,36 @@ object VariantExpressionEvalUtils {
def isValidVariant(input: VariantVal): Boolean =
VariantUtil.isValidVariant(input.getValue, input.getMetadata)
+ /** Throws `INVALID_VARIANT_PATH` on a malformed path or on the empty (root
`$`) path. */
+ def parseVariantDeletePath(pathValue: String): Array[VariantPathSegment] = {
+ val parsed = VariantPathParser.parse(pathValue).getOrElse {
+ throw QueryExecutionErrors.invalidVariantPath(pathValue,
"variant_delete")
+ }
+ if (parsed.isEmpty) {
+ throw QueryExecutionErrors.invalidVariantPath(pathValue,
"variant_delete")
+ }
+ parsed
+ }
+
+ def toJavaSegments(
+ segments: Array[VariantPathSegment]): Array[VariantBuilder.PathSegment]
= {
+ segments.map {
+ case ObjectExtraction(key) => new VariantBuilder.ObjectKeySegment(key)
+ case ArrayExtraction(index) => new
VariantBuilder.ArrayIndexSegment(index)
+ }
+ }
+
+ def deleteAtPath(
+ input: VariantVal,
+ javaSegments: Array[VariantBuilder.PathSegment]): VariantVal = {
+ val v = new Variant(input.getValue, input.getMetadata)
+ val out = VariantBuilder.deleteAtPath(v, javaSegments)
+ new VariantVal(out.getValue, out.getMetadata)
+ }
+
+ def deleteAtPath(input: VariantVal, path: UTF8String): VariantVal =
+ deleteAtPath(input, toJavaSegments(parseVariantDeletePath(path.toString)))
+
/** Cast a Spark value from `dataType` into the variant type. */
def castToVariant(input: Any, dataType: DataType): VariantVal = {
// Enforce strict check because it is illegal for input struct/map/variant
to contain duplicate
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 5d78f11bf86f..0914a8521b1f 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
@@ -662,6 +662,148 @@ object VariantGetExpressionBuilder extends
VariantGetExpressionBuilderBase(true)
// scalastyle:on line.size.limit
object TryVariantGetExpressionBuilder extends
VariantGetExpressionBuilderBase(false)
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = "_FUNC_(v, path1[, path2, ...]) - Removes fields or array elements
from a variant at " +
+ "the given JSONPath locations. Multiple paths are applied left to right.
Returns NULL if " +
+ "`v` is NULL; NULL paths are skipped.",
+ arguments = """
+ Arguments:
+ * v - A variant value to mutate.
+ * path1, path2, ... - One or more string expressions, each evaluating to
a JSONPath
+ identifying a deletion 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.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_(parse_json('{"a": 1, "b": 2, "c": 3, "items": [1, 2,
3]}'), NULL, '$.a', '$.c');
+ {"b":2,"items":[1,2,3]}
+ > SELECT _FUNC_(parse_json('{"a": 1, "b": 2, "c": 3, "items": [1, 2,
3]}'), '$.missing');
+ {"a":1,"b":2,"c":3,"items":[1,2,3]}
+ > SELECT _FUNC_(parse_json('{"a": 1, "b": 2, "c": 3, "items": [1, 2,
3]}'), '$.items[0]', '$.items[0]');
+ {"a":1,"b":2,"c":3,"items":[3]}
+ > SELECT _FUNC_(NULL, '$.a');
+ NULL
+ """,
+ since = "5.0.0",
+ group = "variant_funcs"
+)
+// scalastyle:on line.size.limit
+case class VariantDelete(children: Seq[Expression])
+ extends Expression
+ with ExpectsInputTypes {
+
+ override def dataType: DataType = VariantType
+
+ override def nullable: Boolean = children.headOption.forall(_.nullable)
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ // First argument is the variant; subsequent arguments are JSONPath
strings.
+ VariantType +: Seq.fill(math.max(children.length - 1, 0))(
+ StringTypeWithCollation(supportsTrimCollation = true))
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ if (children.length < 2) {
+ // `wrongNumArgsError` already quotes the function name via `toSQLId`,
so pass the raw name.
+ throw QueryCompilationErrors.wrongNumArgsError(
+ prettyName, Seq("> 1"), children.length)
+ }
+ super.checkInputDataTypes()
+ }
+
+ private def variantChild: Expression = children.head
+ private def pathChildren: Seq[Expression] = children.tail
+
+ @transient private lazy val pathArgs: Seq[VariantDelete.DeletePathArg] =
+ pathChildren.flatMap(VariantDelete.toPathArg)
+
+ override def eval(input: InternalRow): Any = {
+ val inputVariant = variantChild.eval(input).asInstanceOf[VariantVal]
+ if (inputVariant == null) return null
+ var current = inputVariant
+ val args = pathArgs
+ var i = 0
+ while (i < args.length) {
+ args(i) match {
+ case parsed: VariantDelete.ParsedDeletePath =>
+ current = VariantExpressionEvalUtils.deleteAtPath(current,
parsed.javaSegments)
+ case VariantDelete.DynamicDeletePath(expr) =>
+ val pathVal = expr.eval(input).asInstanceOf[UTF8String]
+ if (pathVal != null) {
+ current = VariantExpressionEvalUtils.deleteAtPath(current, pathVal)
+ }
+ }
+ i += 1
+ }
+ current
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = {
+ val cls = VariantExpressionEvalUtils.getClass.getName.stripSuffix("$")
+ val variantValType = CodeGenerator.javaType(VariantType)
+ val childCode = variantChild.genCode(ctx)
+ val current = ctx.freshName("vdCurrent")
+
+ val perPath = pathArgs.map {
+ case parsed: VariantDelete.ParsedDeletePath =>
+ val parsedArg = ctx.addReferenceObj("vdParsed", parsed)
+ s"$current = $cls.deleteAtPath($current, $parsedArg.javaSegments());"
+ case VariantDelete.DynamicDeletePath(expr) =>
+ val pCode = expr.genCode(ctx)
+ s"""
+ |${pCode.code}
+ |if (!${pCode.isNull}) {
+ | $current = $cls.deleteAtPath($current, ${pCode.value});
+ |}
+ """.stripMargin
+ }.mkString("\n")
+
+ val code = code"""
+ ${childCode.code}
+ boolean ${ev.isNull} = ${childCode.isNull};
+ $variantValType ${ev.value} = ${CodeGenerator.defaultValue(VariantType)};
+ if (!${ev.isNull}) {
+ $variantValType $current = ${childCode.value};
+ $perPath
+ ${ev.value} = $current;
+ }
+ """
+ ev.copy(code = code)
+ }
+
+ override def prettyName: String = "variant_delete"
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): VariantDelete = copy(children =
newChildren)
+}
+
+object VariantDelete {
+ sealed trait DeletePathArg
+ case class ParsedDeletePath(segments: Array[VariantPathSegment]) extends
DeletePathArg {
+ // `VariantBuilder.PathSegment` is not `Serializable`, so the cached Java
form is
+ // `@transient` and re-initialized once per executor task after
deserialization.
+ @transient lazy val javaSegments: Array[VariantBuilder.PathSegment] =
+ VariantExpressionEvalUtils.toJavaSegments(segments)
+ }
+ case class DynamicDeletePath(expr: Expression) extends DeletePathArg
+
+ private[variant] def toPathArg(child: Expression): Option[DeletePathArg] = {
+ if (child.foldable) {
+ val v = child.eval()
+ if (v == null) {
+ None
+ } else {
+ Some(ParsedDeletePath(
+
VariantExpressionEvalUtils.parseVariantDeletePath(v.asInstanceOf[UTF8String].toString)))
+ }
+ } else {
+ Some(DynamicDeletePath(child))
+ }
+ }
+}
+
case class VariantExplode(child: Expression) extends UnaryExpression with
Generator
with ExpectsInputTypes {
override def inputTypes: Seq[AbstractDataType] = Seq(VariantType)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 08c9e0462b8b..4b50a90a30f7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -3053,6 +3053,12 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
messageParameters = Map("path" -> path, "functionName" ->
toSQLId(functionName)))
}
+ def invalidVariantPath(path: String, functionName: String): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "INVALID_VARIANT_PATH",
+ messageParameters = Map("path" -> path, "functionName" ->
toSQLId(functionName)))
+ }
+
def malformedVariant(): Throwable = new SparkRuntimeException(
"MALFORMED_VARIANT",
Map.empty
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 dc42cd7fbb2e..467cb6335a2e 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
@@ -24,6 +24,7 @@ import scala.reflect.runtime.universe.TypeTag
import org.apache.spark.{SparkFunSuite, SparkRuntimeException}
import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.ResolveTimeZone
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
@@ -1200,4 +1201,114 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
// Null input.
checkEvaluation(IsValidVariant(Literal.create(null, VariantType)), null)
}
+
+ test("variant_delete") {
+ def checkDelete(input: String, paths: Seq[String], expected: String): Unit
= {
+ val pathLits: Seq[Expression] = paths.map(p => Literal.create(p,
StringType))
+ val expr = VariantDelete(Literal(parseJson(input)) +: pathLits)
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(Cast(expr, StringType)),
+ expected)
+ }
+
+ checkDelete("""{"a": 1, "b": 2}""", Seq("$.a"), """{"b":2}""")
+ checkDelete("""{"a": 1, "b": 2, "c": 3}""", Seq("$.a", "$.c"),
"""{"b":2}""")
+ checkDelete("""{"a": 1}""", Seq("$.missing"), """{"a":1}""")
+ checkDelete("[1, 2, 3]", Seq("$[1]"), "[1,3]")
+ checkDelete("[1, 2, 3]", Seq("$[10]"), "[1,2,3]")
+
+ // Cascading deletes propagate state across paths.
+ checkDelete("[1, 2, 3]", Seq("$[0]", "$[0]", "$[0]"), "[]")
+ checkDelete("""{"a":[1,2,3]}""", Seq("$.a[0]", "$.a[0]", "$.a[0]"),
"""{"a":[]}""")
+
+ checkDelete("""{"a": {"b": 1, "c": 2}}""", Seq("$.a.b"),
"""{"a":{"c":2}}""")
+ checkDelete("""[{"b": 1, "c": 2}]""", Seq("$[0].b"), """[{"c":2}]""")
+
+ // Empty containers are preserved; the parent is never collapsed to NULL.
+ checkDelete("""{"a": 1}""", Seq("$.a"), "{}")
+ checkDelete("[1]", Seq("$[0]"), "[]")
+ checkDelete("""{"a": {"b": 1}}""", Seq("$.a.b"), """{"a":{}}""")
+ checkDelete("""{"a": []}""", Seq("$.a[0]"), """{"a":[]}""")
+
+ checkDelete(
+ """{"a": {"b": {"c": {"d": 1}}}}""",
+ Seq("$.a.b.c.d"),
+ """{"a":{"b":{"c":{}}}}""")
+
+ checkDelete("""{"a": [10, 20, 30]}""", Seq("$.a[1]"), """{"a":[10,30]}""")
+ checkDelete(
+ """{"a": [{"b": 1, "c": 2}, {"b": 3}]}""",
+ Seq("$.a[0].b"),
+ """{"a":[{"c":2},{"b":3}]}""")
+
+ checkDelete("""{"a": 1, "b": 2}""", Seq("$['a']"), """{"b":2}""")
+ checkDelete("""{"a": 1, "b": 2}""", Seq("""$["a"]"""), """{"b":2}""")
+
+ // Pure deep-array nesting: only `ArrayIndexSegment`s, never visits the
`OBJECT` branch.
+ checkDelete("[[[1, 2, 3]]]", Seq("$[0][0][1]"), "[[[1,3]]]")
+ checkDelete("[[10, 20], [30, 40]]", Seq("$[0][1]"), "[[10],[30,40]]")
+
+ // All three key notations (`.k`, `['k']`, `["k"]`) alternating within a
single path.
+ checkDelete(
+ """{"a": {"b": {"c": 1, "d": 2}}}""",
+ Seq("""$['a'].b["c"]"""),
+ """{"a":{"b":{"d":2}}}""")
+
+ checkDelete("""{"": 1, "a": 2}""", Seq("$['']"), """{"a":2}""")
+ checkDelete("""{"?": 1, "a": 2}""", Seq("$['?']"), """{"a":2}""")
+ checkDelete(
+ """{"key with spaces": 1, "a": 2}""", Seq("$['key with spaces']"),
"""{"a":2}""")
+ checkDelete("""{"fb:testid": 1, "a": 2}""", Seq("$.fb:testid"),
"""{"a":2}""")
+
+ checkDelete("""{"a": 1, "b": 2}""", Seq(null, "$.a"), """{"b":2}""")
+
+ // After a deletion empties the parent, a subsequent nested path is a
silent no-op.
+ checkDelete("""{"a": {"b": 1}}""", Seq("$.a", "$.a.b"), "{}")
+
+ // Type mismatches between segment and value are silent no-ops.
+ checkDelete("""{"a": 5}""", Seq("$.a.b"), """{"a":5}""")
+ checkDelete("[1, 2, 3]", Seq("$.a"), "[1,2,3]")
+ checkDelete("""{"a": 1}""", Seq("$[0]"), """{"a":1}""")
+
+ checkDelete("""{"a": 1, "b": 2}""", Seq[String](null), """{"a":1,"b":2}""")
+
+ // All literal-NULL paths: `flatMap` leaves `pathArgs` empty; input is
returned unchanged.
+ checkDelete("""{"a": 1, "b": 2}""", Seq(null, null, null),
"""{"a":1,"b":2}""")
+
+ checkDelete("""{"a": null, "b": 2}""", Seq("$.a"), """{"b":2}""")
+ checkDelete("[null, 1, null]", Seq("$[0]"), "[1,null]")
+
+ // Mixed literal + dynamic path exercises both `ParsedDeletePath` and
`DynamicDeletePath`
+ // arms of `eval` in a single call.
+ val mixedLitDyn = VariantDelete(Seq(
+ Literal(parseJson("""{"a": 1, "b": 2, "c": 3}""")),
+ Literal("$.a"),
+ BoundReference(0, StringType, nullable = true)))
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(Cast(mixedLitDyn, StringType)),
+ """{"b":2}""",
+ InternalRow(UTF8String.fromString("$.c")))
+
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(
+ Cast(VariantDelete(Seq(Literal.create(null, VariantType),
Literal("$.a"))), StringType)),
+ null)
+
+ checkErrorInExpression[SparkRuntimeException](
+ ResolveTimeZone.resolveTimeZones(
+ VariantDelete(Seq(Literal(parseJson("""{"a": 1}""")), Literal("$")))),
+ "INVALID_VARIANT_PATH",
+ Map("path" -> "$", "functionName" -> "`variant_delete`"))
+
+ checkErrorInExpression[SparkRuntimeException](
+ ResolveTimeZone.resolveTimeZones(
+ VariantDelete(Seq(Literal(parseJson("""{"a": 1}""")), Literal(".a")))),
+ "INVALID_VARIANT_PATH",
+ Map("path" -> ".a", "functionName" -> "`variant_delete`"))
+
+ val noPaths = VariantDelete(Seq(Literal(parseJson("""{"a": 1}"""))))
+ intercept[org.apache.spark.sql.AnalysisException] {
+ noPaths.checkInputDataTypes()
+ }
+ }
}
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 33aff976434c..08b28e2b2081 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
@@ -2759,6 +2759,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite
with Logging {
fn.is_valid_variant(fn.parse_json(fn.col("g")))
}
+ functionTest("variant_delete") {
+ fn.variant_delete(fn.parse_json(fn.col("g")), "$.a", "$.b")
+ }
+
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_variant_delete.explain
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_delete.explain
new file mode 100644
index 000000000000..6b54167b66d6
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_delete.explain
@@ -0,0 +1,2 @@
+Project
[variant_delete(static_invoke(VariantExpressionEvalUtils.parseJson(g#0, false,
true, true)), $.a, $.b) AS variant_delete(parse_json(g), $.a, $.b)#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_variant_delete.json
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_delete.json
new file mode 100644
index 000000000000..d225f18a73c2
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_delete.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": "variant_delete",
+ "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": "variant_delete",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass":
"org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "literal": {
+ "string": "$.b"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "variant_delete",
+ "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": "variant_delete",
+ "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_variant_delete.proto.bin
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_delete.proto.bin
new file mode 100644
index 000000000000..12f3535cbb6b
Binary files /dev/null and
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_delete.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 be8559d16cb7..6003c7af52d3 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
@@ -552,6 +552,7 @@
| 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.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.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> |
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 a58d701ea0c9..787292f55b5a 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
@@ -196,6 +196,105 @@ class VariantSuite extends SharedSparkSession with
ExpressionEvalHelper {
}
}
+ test("variant_delete with literal paths") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_delete(parse_json('{\"a\": 1, \"b\": 2,
\"c\": 3}'), " +
+ "'$.a', '$.missing', '$.c'))"),
+ rows("""{"b":2}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_delete(parse_json('{\"a\": 1, \"b\": 2}'),
NULL, '$.a'))"),
+ rows("""{"b":2}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_delete(parse_json('[1, 2, 3]'), '$[1]'))"),
+ rows("[1,3]"))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_delete(CAST(NULL AS VARIANT), '$.a'))"),
+ rows(null))
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT variant_delete(parse_json('{\"a\": 1}'), '$')").collect()
+ },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "$", "functionName" ->
toSQLId("variant_delete")))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT variant_delete(parse_json('{}'))").collect()
+ },
+ condition = "WRONG_NUM_ARGS.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "functionName" -> toSQLId("variant_delete"),
+ "expectedNum" -> "> 1",
+ "actualNum" -> "1",
+ "docroot" -> "https://spark.apache.org/docs/latest";))
+ }
+
+ test("non-literal variant_delete") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+ val df = Seq(
+ ("""{"a": 1, "b": 2}""", "$.a", null, 2),
+ ("""{"a": 1, "b": 2, "c": 3}""", "$.a", "$.c", 2),
+ ("""{"a": 1}""", "$.missing", null, 2),
+ ("""{"a": 1, "b": 2}""", null, null, 2),
+ (null, "$.a", null, 2),
+ ("""{"a": 1}""", "$", null, 1),
+ ("""{"a": 1}""", "abc", null, 0)).toDF("json", "path", "path2", "valid")
+ val v = parse_json(col("json"))
+ val df1 = df.where($"valid" === 2).select(
+ to_json(variant_delete(v, col("path"), col("path2"))))
+ checkAnswer(
+ df1,
+ rows("""{"b":2}""", """{"b":2}""", """{"a":1}""", """{"a":1,"b":2}""",
null))
+
+ val dfInvalidRoot = df.where($"valid" === 1).select(
+ to_json(variant_delete(v, col("path"))))
+ checkError(
+ exception = intercept[SparkRuntimeException] { dfInvalidRoot.collect() },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "$", "functionName" ->
toSQLId("variant_delete")))
+
+ val df2 = df.where($"valid" === 0).select(to_json(variant_delete(v,
col("path"))))
+ checkError(
+ exception = intercept[SparkRuntimeException] { df2.collect() },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "abc", "functionName" ->
toSQLId("variant_delete")))
+
+ withTable("t") {
+ df.withColumn("v", parse_json(col("json"))).write.saveAsTable("t")
+ checkAnswer(
+ sql("select to_json(variant_delete(v, path, path2)) from t where valid
= 2"),
+ rows("""{"b":2}""", """{"b":2}""", """{"a":1}""", """{"a":1,"b":2}""",
null))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("select to_json(variant_delete(v, path)) from t where valid =
0").collect()
+ },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "abc", "functionName" ->
toSQLId("variant_delete")))
+ }
+ }
+
+ test("variant_delete mixed literal and dynamic paths") {
+ 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": 2, "c": 3}""", "$.c"),
+ ("""{"a": 1, "b": 2}""", "$.b"),
+ (null, "$.a")
+ ).toDF("json", "path")
+ val v = parse_json(col("json"))
+ val out = df.select(to_json(variant_delete(v, lit("$.a"),
col("path"))).alias("r"))
+ checkAnswer(out, rows("""{"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]
