This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new a5dbd23631cc [SPARK-58062][SQL] Add `variant_strip_nulls` expression
a5dbd23631cc is described below
commit a5dbd23631cc53605d200d7b89c92fe2bf48e0f9
Author: bojana-db <[email protected]>
AuthorDate: Fri Aug 14 15:57:46 2026 +0000
[SPARK-58062][SQL] Add `variant_strip_nulls` expression
### What changes were proposed in this pull request?
Adds the SQL function` variant_strip_nulls(v[, includeArrays])`, which
recursively removes Variant null fields from objects (and, by default, Variant
null elements from arrays) in a Variant value.
Details:
- Object path: a field whose value is a Variant null (the JSON null
literal) is removed entirely; recursion walks nested objects and arrays at all
levels;
- includeArrays (BOOLEAN, default true): when true, Variant null elements
are also removed from arrays; when false, array null elements are kept, but
null fields of objects nested anywhere (including inside arrays) are still
stripped;
- Empty containers are preserved: an object/array emptied by stripping
stays {} / []; the parent is never collapsed to NULL;
- Non-container inputs — scalars and a top-level Variant null
(parse_json('null')) — are returned unchanged;
- NULL intolerant: any NULL argument returns NULL.
### Why are the changes needed?
There is no built-in way to drop Variant null entries from a Variant; today
users must convert it to another type (e.g. a map) and back.
### Does this PR introduce _any_ user-facing change?
Yes, a new SQL function (and Scala/Python `functions` API.
### 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 #57933 from bojana-db/backport-SPARK-58062-4.3.
Authored-by: bojana-db <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../apache/spark/types/variant/VariantBuilder.java | 59 ++++++++++++
.../source/reference/pyspark.sql/functions.rst | 1 +
python/pyspark/sql/connect/functions/builtin.py | 7 ++
python/pyspark/sql/functions/__init__.py | 1 +
python/pyspark/sql/functions/builtin.py | 46 +++++++++
python/pyspark/sql/tests/test_functions.py | 9 ++
.../scala/org/apache/spark/sql/functions.scala | 26 ++++++
.../sql/catalyst/analysis/FunctionRegistry.scala | 1 +
.../variant/VariantExpressionEvalUtils.scala | 6 ++
.../expressions/variant/variantExpressions.scala | 91 +++++++++++++++++-
.../variant/VariantExpressionSuite.scala | 72 +++++++++++++-
.../apache/spark/sql/PlanGenerationTestSuite.scala | 4 +
.../function_variant_strip_nulls.explain | 2 +
.../queries/function_variant_strip_nulls.json | 104 +++++++++++++++++++++
.../queries/function_variant_strip_nulls.proto.bin | Bin 0 -> 967 bytes
.../sql-functions/sql-expression-schema.md | 1 +
.../scala/org/apache/spark/sql/VariantSuite.scala | 76 +++++++++++++++
17 files changed, 503 insertions(+), 3 deletions(-)
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 296bc339ee08..d9eed06bf3a6 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
@@ -171,6 +171,17 @@ public class VariantBuilder {
return builder.result();
}
+ // Return a new variant with null-valued object fields removed, recursing
into nested objects
+ // and arrays. When `includeArrays` is true, null array elements are removed
too; when false,
+ // arrays keep their nulls but objects inside them are still cleaned. A
container emptied by
+ // stripping is preserved as {} / [] rather than collapsed to a variant
null, and a top-level
+ // variant null is returned unchanged. The result is always rebuilt with
fresh metadata.
+ public static Variant stripNulls(Variant v, boolean includeArrays) {
+ VariantBuilder builder = new VariantBuilder(false);
+ builder.appendWithNullStrippingImpl(v.value, v.metadata, v.pos,
includeArrays);
+ return builder.result();
+ }
+
// Build the variant metadata from `dictionaryKeys` and return the variant
result.
public Variant result() {
int numKeys = dictionaryKeys.size();
@@ -899,6 +910,54 @@ public class VariantBuilder {
}
}
+ private void appendWithNullStrippingImpl(
+ byte[] value, byte[] metadata, int pos, boolean includeArrays) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ if (basicType == OBJECT) {
+ 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;
+ // Drop the whole field when its value is a variant null.
+ if (getType(value, elementPos) == Type.NULL) {
+ continue;
+ }
+ String key = getMetadataKey(metadata, id);
+ int newId = addKey(key);
+ fields.add(new FieldEntry(key, newId, writePos - start));
+ appendWithNullStrippingImpl(value, metadata, elementPos,
includeArrays);
+ }
+ finishWritingObject(start, fields);
+ return null;
+ });
+ } else if (basicType == ARRAY) {
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ ArrayList<Integer> offsets = new ArrayList<>(size);
+ int start = writePos;
+ for (int i = 0; i < size; ++i) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ // Drop variant-null elements only when stripping arrays; otherwise
keep them but still
+ // recurse into nested containers.
+ if (includeArrays && getType(value, elementPos) == Type.NULL) {
+ continue;
+ }
+ offsets.add(writePos - start);
+ appendWithNullStrippingImpl(value, metadata, elementPos,
includeArrays);
+ }
+ finishWritingArray(start, offsets);
+ return null;
+ });
+ } else {
+ // Scalars and standalone variant nulls are appended 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 cc1756304549..5147e6c90ba6 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -607,6 +607,7 @@ VARIANT Functions
variant_array_append
try_variant_array_append
variant_delete
+ variant_strip_nulls
variant_get
variant_insert
try_variant_insert
diff --git a/python/pyspark/sql/connect/functions/builtin.py
b/python/pyspark/sql/connect/functions/builtin.py
index 0713eecbe9b0..77e8889f4114 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -2294,6 +2294,13 @@ def try_variant_array_append(
try_variant_array_append.__doc__ =
pysparkfuncs.try_variant_array_append.__doc__
+def variant_strip_nulls(v: "ColumnOrName", include_arrays: bool = True) ->
Column:
+ return _invoke_function("variant_strip_nulls", _to_col(v),
lit(include_arrays))
+
+
+variant_strip_nulls.__doc__ = pysparkfuncs.variant_strip_nulls.__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 0e2c6386a4d4..2ed8438ac8b5 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -486,6 +486,7 @@ __all__ = [ # noqa: F405
"variant_array_append",
"try_variant_array_append",
"variant_delete",
+ "variant_strip_nulls",
"variant_get",
"variant_insert",
"try_variant_insert",
diff --git a/python/pyspark/sql/functions/builtin.py
b/python/pyspark/sql/functions/builtin.py
index 6ed9a1049858..89074046dcf4 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -23187,6 +23187,52 @@ def try_variant_array_append(
)
+@_try_remote_functions
+def variant_strip_nulls(v: "ColumnOrName", include_arrays: bool = True) ->
Column:
+ """
+ Recursively removes object fields and array elements whose value is a
variant null, unless
+ `include_arrays` is False, in which case null array elements are kept.
Returns NULL if any
+ argument is NULL.
+
+ .. versionadded:: 4.3.0
+
+ Parameters
+ ----------
+ v : :class:`~pyspark.sql.Column` or str
+ a variant column or column name
+ include_arrays : bool, optional
+ whether null elements are also removed from arrays (default True).
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ a variant column with variant null fields/elements removed
+
+ Examples
+ --------
+ >>> from pyspark.sql.functions import lit, parse_json, to_json,
variant_strip_nulls
+ >>> df = spark.createDataFrame([{
+ ... 'json': '''{ "a" : 1, "b" : null, "c" : [1, null], "d" : { "e" :
null, "f" : 4 } }'''
+ ... }])
+ >>> v = parse_json(df.json)
+ >>> df.select(to_json(variant_strip_nulls(v)).alias("r")).collect()
+ [Row(r='{"a":1,"c":[1],"d":{"f":4}}')]
+ >>> df.select(to_json(variant_strip_nulls(v, False)).alias("r")).collect()
+ [Row(r='{"a":1,"c":[1,null],"d":{"f":4}}')]
+ >>> df.select(variant_strip_nulls(lit(None)).alias("r")).collect()
+ [Row(r=None)]
+ >>> df2 = spark.createDataFrame([{'json': '{"a": null}'}, {'json':
'null'}])
+ >>> v2 = parse_json(df2.json)
+ >>> df2.select(to_json(variant_strip_nulls(v2)).alias("r")).collect()
+ [Row(r='{}'), Row(r='null')]
+ """
+ from pyspark.sql.classic.column import _to_java_column
+
+ return _invoke_function(
+ "variant_strip_nulls", _to_java_column(v),
_enum_to_value(include_arrays)
+ )
+
+
@_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 41764083c3c8..87be88585965 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -3641,6 +3641,15 @@ class FunctionsTestsMixin:
df.select(F.to_json(F.try_variant_array_append(arr, df.arrpath,
F.lit(9)))),
["[1,2,9]", "[[3,9],4]"],
)
+ strip_v = F.parse_json(F.lit('{"a": 1, "b": null, "c": [1, null]}'))
+ check(
+ df.select(F.to_json(F.variant_strip_nulls(strip_v))),
+ ['{"a":1,"c":[1]}', '{"a":1,"c":[1]}'],
+ )
+ check(
+ df.select(F.to_json(F.variant_strip_nulls(strip_v, False))),
+ ['{"a":1,"c":[1,null]}', '{"a":1,"c":[1,null]}'],
+ )
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 0cc822c2ec01..8d9011c7eba4 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
@@ -14555,6 +14555,32 @@ object functions {
def try_variant_array_append(v: Column, path: String, value: Column): Column
=
Column.fn("try_variant_array_append", v, lit(path), value)
+ /**
+ * Recursively removes object fields and array elements whose value is a
variant null. Returns
+ * NULL if `v` is NULL.
+ *
+ * @param v
+ * a variant column.
+ * @group variant_funcs
+ * @since 4.3.0
+ */
+ def variant_strip_nulls(v: Column): Column =
Column.fn("variant_strip_nulls", v)
+
+ /**
+ * Recursively removes object fields and array elements whose value is a
variant null, unless
+ * `includeArrays` is false, in which case null array elements are kept.
Returns NULL if any
+ * argument is NULL.
+ *
+ * @param v
+ * a variant column.
+ * @param includeArrays
+ * whether null elements are also removed from arrays.
+ * @group variant_funcs
+ * @since 4.3.0
+ */
+ def variant_strip_nulls(v: Column, includeArrays: Boolean): Column =
+ Column.fn("variant_strip_nulls", v, lit(includeArrays))
+
/**
* 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 e8bc63914716..81bdde42f608 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
@@ -996,6 +996,7 @@ object FunctionRegistry {
expressionBuilder("try_variant_set", TryVariantSetExpressionBuilder),
expressionBuilder("variant_array_append",
VariantArrayAppendExpressionBuilder),
expressionBuilder("try_variant_array_append",
TryVariantArrayAppendExpressionBuilder),
+ expressionBuilder("variant_strip_nulls",
VariantStripNullsExpressionBuilder),
// 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 da1bfbcea829..354c861830a1 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
@@ -270,6 +270,12 @@ object VariantExpressionEvalUtils {
arrayAppendAtPath(input, javaSegments, pathStr, value, valueDataType,
functionName, failOnError)
}
+ def stripNulls(input: VariantVal, includeArrays: Boolean): VariantVal = {
+ val v = new Variant(input.getValue, input.getMetadata)
+ val out = VariantBuilder.stripNulls(v, includeArrays)
+ new VariantVal(out.getValue, out.getMetadata)
+ }
+
/** 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 723af5a4fe92..0d29f243a0b9 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
@@ -32,8 +32,8 @@ import
org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.json.JsonInferSchema
import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature,
InputParameter}
+import org.apache.spark.sql.catalyst.trees.{BinaryLike, UnaryLike}
import org.apache.spark.sql.catalyst.trees.TreePattern.{TreePattern,
VARIANT_GET}
-import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData,
GenericArrayData, QuotingUtils}
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase,
QueryExecutionErrors}
@@ -1496,7 +1496,94 @@ object VariantArrayAppendExpressionBuilder extends
VariantArrayAppendExpressionB
)
// scalastyle:on line.size.limit
object TryVariantArrayAppendExpressionBuilder
- extends VariantArrayAppendExpressionBuilderBase(false)
+extends VariantArrayAppendExpressionBuilderBase(false)
+
+case class VariantStripNulls(child: Expression, includeArrays: Expression)
+ extends RuntimeReplaceable
+ with ExpectsInputTypes
+ with BinaryLike[Expression]
+ with QueryErrorsBase {
+
+ override def left: Expression = child
+ override def right: Expression = includeArrays
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(VariantType,
BooleanType)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val result = super.checkInputDataTypes()
+ if (result.isFailure) {
+ result
+ } else if (!includeArrays.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> toSQLId("include_arrays"),
+ "inputType" -> toSQLType(includeArrays.dataType),
+ "inputExpr" -> toSQLExpr(includeArrays)))
+ } else {
+ TypeCheckResult.TypeCheckSuccess
+ }
+ }
+
+ override lazy val replacement: Expression = StaticInvoke(
+ VariantExpressionEvalUtils.getClass,
+ VariantType,
+ "stripNulls",
+ Seq(child, includeArrays),
+ inputTypes,
+ returnNullable = false)
+
+ override def prettyName: String = "variant_strip_nulls"
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression, newRight: Expression): VariantStripNulls =
+ copy(child = newLeft, includeArrays = newRight)
+}
+
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+ usage = "_FUNC_(v[, include_arrays]) - Recursively removes object fields and
array elements " +
+ "whose value is a variant null, unless `include_arrays` is false, in which
case null array " +
+ "elements are kept. Returns NULL if any argument is NULL.",
+ arguments = """
+ Arguments:
+ * v - The variant value to strip.
+ * include_arrays - An optional boolean (default true). Must be a
constant.
+ """,
+ examples = """
+ Examples:
+ > SELECT _FUNC_(parse_json('{"a": 1, "b": null, "c": 3}'));
+ {"a":1,"c":3}
+ > SELECT _FUNC_(parse_json('[1, null, 3]'));
+ [1,3]
+ > SELECT _FUNC_(parse_json('{"a": {"b": null, "c": [1, null]}}'));
+ {"a":{"c":[1]}}
+ > SELECT _FUNC_(parse_json('{"a": [1, null], "b": null}'), false);
+ {"a":[1,null]}
+ > SELECT _FUNC_(parse_json('{"a": null}'));
+ {}
+ > SELECT _FUNC_(parse_json('null'));
+ null
+ > SELECT _FUNC_(NULL);
+ NULL
+ """,
+ since = "4.3.0",
+ group = "variant_funcs"
+)
+// scalastyle:on line.size.limit
+object VariantStripNullsExpressionBuilder extends ExpressionBuilder {
+ override def functionSignature: Option[FunctionSignature] = {
+ val vArg = InputParameter("v")
+ val includeArraysArg =
+ InputParameter("include_arrays", Some(Literal.create(true, BooleanType)))
+ Some(FunctionSignature(Seq(vArg, includeArraysArg)))
+ }
+
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression = {
+ assert(expressions.size == 2)
+ VariantStripNulls(expressions(0), expressions(1))
+ }
+}
case class VariantExplode(child: Expression) extends UnaryExpression with
Generator
with ExpectsInputTypes {
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 b3248609ae3e..421ef186beba 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
@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeConstants._
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-import org.apache.spark.types.variant.VariantBuilder
+import org.apache.spark.types.variant.{Variant, VariantBuilder}
import org.apache.spark.types.variant.VariantUtil._
import org.apache.spark.unsafe.types.{UTF8String, VariantVal}
import org.apache.spark.util.collection.Utils.createArray
@@ -1893,4 +1893,74 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
"VARIANT_SIZE_LIMIT",
name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`"))
}
+
+ test("variant_strip_nulls") {
+ // Strip `input`, render the result back to JSON, and compare.
`includeArrays` defaults to true.
+ def check(input: String, expected: String, includeArrays: Boolean = true):
Unit = {
+ val expr = VariantStripNulls(Literal(parseJson(input)),
Literal(includeArrays))
+ val result = replace(expr).eval().asInstanceOf[VariantVal]
+ val json = if (result == null) null
+ else new Variant(result.getValue,
result.getMetadata).toJson(ZoneOffset.UTC)
+ assert(json == expected)
+ }
+
+ // The optional `include_arrays` argument defaults to true in the function
signature.
+
assert(VariantStripNullsExpressionBuilder.functionSignature.get.parameters.last.default
+ .contains(Literal.create(true, BooleanType)))
+
+ // include_arrays must be a constant; a non-foldable expression is
rejected.
+ assert(VariantStripNulls(
+ Literal(parseJson("[1, null]")), BoundReference(0, BooleanType, nullable
= true))
+ .checkInputDataTypes().isFailure)
+
+ check("""{"a": 1, "b": null, "c": 3}""", """{"a":1,"c":3}""")
+ check("[1, null, 3]", "[1,3]")
+ check("""{"user": {"name": "Alice", "age": null}}""",
"""{"user":{"name":"Alice"}}""")
+ check("""{"a": [1, null, {"b": null, "c": 2}]}""", """{"a":[1,{"c":2}]}""")
+ check("[[1, null], [null]]", "[[1],[]]")
+
+ // Empty containers are preserved; the parent is never collapsed.
+ check("""{"a": null}""", "{}")
+ check("[null]", "[]")
+ check("""{"a": {"b": null}}""", """{"a":{}}""")
+ check("""{"a": [null]}""", """{"a":[]}""")
+ check("{}", "{}")
+ check("[]", "[]")
+
+ // Top-level variant null and scalars are returned unchanged.
+ check("null", "null")
+ check("42", "42")
+ check("\"hi\"", "\"hi\"")
+
+ check("""{"a": {"b": {"c": null, "d": 4}}}""", """{"a":{"b":{"d":4}}}""")
+
+ check("""{"a": 300, "b": null, "c": 100000, "d": 10000000000}""",
+ """{"a":300,"c":100000,"d":10000000000}""")
+ check("""[1000000, null, "hello world", null, 10000000000]""",
+ """[1000000,"hello world",10000000000]""")
+ val bigStr = "x".repeat(300)
+ check(s"""{"k": "$bigStr", "n": null, "m": 3}""",
s"""{"k":"$bigStr","m":3}""")
+ check(s"""[null, "$bigStr", null, 100000]""", s"""["$bigStr",100000]""")
+
+ // `includeArrays = false`.
+ check("""{"a": [1, null, 3], "b": null}""", """{"a":[1,null,3]}""",
includeArrays = false)
+ check(
+ """[{"a": 1, "b": null}, null, {"c": null, "d": 4}]""",
+ """[{"a":1},null,{"d":4}]""",
+ includeArrays = false)
+ // `includeArrays = true` (explicit) strips array nulls.
+ check("""{"a": [1, null, 3]}""", """{"a":[1,3]}""", includeArrays = true)
+
+ // SQL NULL variant input yields SQL NULL.
+ checkEvaluation(
+ Cast(VariantStripNulls(Literal.create(null, VariantType),
Literal(true)), StringType),
+ null)
+ // NULL `includeArrays` yields SQL NULL (the expression is null
intolerant).
+ checkEvaluation(
+ Cast(
+ VariantStripNulls(
+ Literal(parseJson("""{"a": null}""")), Literal.create(null,
BooleanType)),
+ StringType),
+ null)
+ }
}
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 446efbf0c955..5c27902918b1 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
@@ -2803,6 +2803,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite
with Logging {
fn.try_variant_array_append(fn.parse_json(fn.col("g")), "$.a", fn.lit(1))
}
+ functionTest("variant_strip_nulls") {
+ fn.variant_strip_nulls(fn.parse_json(fn.col("g")), false)
+ }
+
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_strip_nulls.explain
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_strip_nulls.explain
new file mode 100644
index 000000000000..f6d8330de828
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_variant_strip_nulls.explain
@@ -0,0 +1,2 @@
+Project
[static_invoke(VariantExpressionEvalUtils.stripNulls(static_invoke(VariantExpressionEvalUtils.parseJson(g#0,
false, true, true)), false)) AS variant_strip_nulls(parse_json(g), false)#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_strip_nulls.json
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_strip_nulls.json
new file mode 100644
index 000000000000..19a249ce8188
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_strip_nulls.json
@@ -0,0 +1,104 @@
+{
+ "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_strip_nulls",
+ "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": {
+ "boolean": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "variant_strip_nulls",
+ "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_strip_nulls",
+ "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_strip_nulls.proto.bin
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_strip_nulls.proto.bin
new file mode 100644
index 000000000000..e1a82aceffcb
Binary files /dev/null and
b/sql/connect/common/src/test/resources/query-tests/queries/function_variant_strip_nulls.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 815408f225e7..05924e1641ca 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
@@ -567,6 +567,7 @@
|
org.apache.spark.sql.catalyst.expressions.variant.VariantGetExpressionBuilder |
variant_get | SELECT variant_get(parse_json('{"a": 1}'), '$.a', 'int') |
struct<variant_get(parse_json({"a": 1}), $.a):int> |
|
org.apache.spark.sql.catalyst.expressions.variant.VariantInsertExpressionBuilder
| variant_insert | SELECT variant_insert(parse_json('{"a": 1}'), '$.b', 2) |
struct<variant_insert(parse_json({"a": 1}), $.b, 2):variant> |
|
org.apache.spark.sql.catalyst.expressions.variant.VariantSetExpressionBuilder |
variant_set | SELECT variant_set(parse_json('{"a": 1}'), '$.a', 2) |
struct<variant_set(parse_json({"a": 1}), $.a, 2, true):variant> |
+|
org.apache.spark.sql.catalyst.expressions.variant.VariantStripNullsExpressionBuilder
| variant_strip_nulls | SELECT variant_strip_nulls(parse_json('{"a": 1, "b":
null, "c": 3}')) | struct<variant_strip_nulls(parse_json({"a": 1, "b": null,
"c": 3}), true):variant> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathBoolean | xpath_boolean |
SELECT xpath_boolean('<a><b>1</b></a>','a/b') |
struct<xpath_boolean(<a><b>1</b></a>, a/b):boolean> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_double |
SELECT xpath_double('<a><b>1</b><b>2</b></a>', 'sum(a/b)') |
struct<xpath_double(<a><b>1</b><b>2</b></a>, sum(a/b)):double> |
| org.apache.spark.sql.catalyst.expressions.xml.XPathDouble | xpath_number |
SELECT xpath_number('<a><b>1</b><b>2</b></a>', 'sum(a/b)') |
struct<xpath_number(<a><b>1</b><b>2</b></a>, sum(a/b)):double> |
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala
index 0c44468b2d89..9c2bfaca03a2 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
@@ -685,6 +685,82 @@ class VariantSuite extends SharedSparkSession with
ExpressionEvalHelper {
}
}
+ test("variant_strip_nulls with literal arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('{\"a\": 1, \"b\":
null, \"c\": 3}')))"),
+ rows("""{"a":1,"c":3}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('[1, null, 3]')))"),
+ rows("[1,3]"))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json(" +
+ "'{\"a\": [null, 3, {\"b\": null, \"c\": [null, 1]}], \"d\": null, " +
+ "\"e\": {\"f\": null, \"g\": 2}}')))"),
+ rows("""{"a":[3,{"c":[1]}],"e":{"g":2}}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json(" +
+ "'{\"a\": 100000, \"b\": null, \"c\": 10000000000, \"d\": \"hello
world\"}')))"),
+ rows("""{"a":100000,"c":10000000000,"d":"hello world"}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('{\"a\": [1, null],
\"b\": null}'), " +
+ "include_arrays => false))"),
+ rows("""{"a":[1,null]}"""))
+
+ // include_arrays = false keeps array null elements but still strips null
fields of objects.
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(" +
+ "parse_json('[{\"a\": 1, \"b\": null}, null, {\"c\": null, \"d\":
4}]'), false))"),
+ rows("""[{"a":1},null,{"d":4}]"""))
+
+ // Empty containers are preserved.
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('{\"a\": null}')))"),
+ rows("{}"))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('[null, null]')))"),
+ rows("[]"))
+
+ // Top-level variant null is unchanged.
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(parse_json('null')))"),
+ rows("null"))
+
+ checkAnswer(
+ sql("SELECT to_json(variant_strip_nulls(CAST(NULL AS VARIANT)))"),
+ rows(null))
+
+ assert(intercept[AnalysisException] {
+ sql("SELECT variant_strip_nulls(parse_json('{\"a\": null}'), c) FROM
VALUES (true) AS t(c)")
+ }.getCondition == "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
+ }
+
+ test("variant_strip_nulls with dynamic arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+ val df = Seq(
+ """{"a": [1, null], "b": null}""",
+ """{"x": null, "y": 2}""",
+ null
+ ).toDF("json")
+ val v = parse_json(col("json"))
+
+ // Single-argument overload defaults include_arrays to true.
+ checkAnswer(
+ df.select(to_json(variant_strip_nulls(v)).alias("r")),
+ rows("""{"a":[1]}""", """{"y":2}""", null))
+
+ // Boolean overload with include_arrays = false preserves array null
elements.
+ checkAnswer(
+ df.select(to_json(variant_strip_nulls(v, false)).alias("r")),
+ rows("""{"a":[1,null]}""", """{"y":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]
