You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/05/05 15:04:52 UTC
(spark) branch branch-4.2 updated: [SPARK-52857] Add is_valid_variant expression
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 805fca60dfbe [SPARK-52857] Add is_valid_variant expression
805fca60dfbe is described below
commit 805fca60dfbe5ecad6f7ace564ffe70bffe17813
Author: chenhao-db <ch...@databricks.com>
AuthorDate: Tue May 5 23:03:30 2026 +0800
[SPARK-52857] Add is_valid_variant expression
### What changes were proposed in this pull request?
Add a new expression `is_valid_variant`. Returns true if the variant is valid, false if it is malformed.
### Why are the changes needed?
This is useful when the data source is not completely known. Other variant expressions will throw a `MALFORMED_VARIANT` exception if any row is malformed, which can be inconvenient to use when only a subset of rows contains malformed variant. This new expression will return a boolean instead of throwing exceptions.
### Does this PR introduce _any_ user-facing change?
Yes. A new SQL expression is added.
### How was this patch tested?
Unit tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #55599 from chenhao-db/is_valid_variant.
Authored-by: chenhao-db <ch...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 109a1309b52d9aef4a0984e0f83297ecec3a708a)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../apache/spark/types/variant/VariantUtil.java | 89 ++++++++++++-
.../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 | 29 ++++
python/pyspark/sql/tests/test_functions.py | 1 +
.../scala/org/apache/spark/sql/functions.scala | 11 ++
.../sql/catalyst/analysis/FunctionRegistry.scala | 1 +
.../variant/VariantExpressionEvalUtils.scala | 3 +
.../expressions/variant/variantExpressions.scala | 32 +++++
.../variant/VariantExpressionSuite.scala | 147 +++++++++++++++++++++
.../apache/spark/sql/PlanGenerationTestSuite.scala | 4 +
.../function_is_valid_variant.explain | 2 +
.../queries/function_is_valid_variant.json | 83 ++++++++++++
.../queries/function_is_valid_variant.proto.bin | Bin 0 -> 761 bytes
.../sql-functions/sql-expression-schema.md | 1 +
16 files changed, 411 insertions(+), 1 deletion(-)
diff --git a/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java b/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java
index 795d46ec2062..ac93246991c0 100644
--- a/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java
+++ b/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java
@@ -428,7 +428,8 @@ public class VariantUtil {
// Get a decimal value from variant value `value[pos...]`.
// Throw `MALFORMED_VARIANT` if the variant is malformed.
public static BigDecimal getDecimalWithOriginalScale(byte[] value, int pos) {
- checkIndex(pos, value.length);
+ // Decimal should at least have header + scale.
+ checkIndex(pos + 1, value.length);
int basicType = value[pos] & BASIC_TYPE_MASK;
int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
if (basicType != PRIMITIVE) throw unexpectedType(Type.DECIMAL);
@@ -589,6 +590,92 @@ public class VariantUtil {
return handler.apply(size, offsetSize, offsetStart, dataStart);
}
+ // Validate whether a variant is well-formed. Returns true if the variant binary is structurally
+ // well-formed (all bounds and type-info checks pass), false if it is malformed.
+ //
+ // This is close to, but not strictly equivalent to, "`toJson` does not throw": this function
+ // does not enforce the `SIZE_LIMIT` check that the `Variant` constructor applies (which throws
+ // `VARIANT_CONSTRUCTOR_SIZE_LIMIT`). The implementation otherwise has the same structure as
+ // `toJson` (see `Variant.toJsonImpl`).
+ //
+ // Implementation note: this `try { ... } catch (SparkRuntimeException e)` is sound only because
+ // every helper invoked by `validateImpl` throws `MALFORMED_VARIANT` /
+ // `UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT` rather than a raw `ArrayIndexOutOfBoundsException` on
+ // malformed input. Preserve that invariant when adding new cases.
+ public static boolean isValidVariant(byte[] value, byte[] metadata) {
+ if (value == null || metadata == null) return false;
+ // Validate the metadata version, similar to the check in the `Variant` constructor.
+ if (metadata.length < 1 || (metadata[0] & VERSION_MASK) != VERSION) return false;
+ try {
+ validateImpl(value, metadata, 0);
+ return true;
+ } catch (SparkRuntimeException e) {
+ return false;
+ }
+ }
+
+ private static void validateImpl(byte[] value, byte[] metadata, int pos) {
+ switch (getType(value, pos)) {
+ case OBJECT:
+ handleObject(value, pos, (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+ 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;
+ getMetadataKey(metadata, id);
+ validateImpl(value, metadata, elementPos);
+ }
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ for (int i = 0; i < size; ++i) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i, offsetSize);
+ int elementPos = dataStart + offset;
+ validateImpl(value, metadata, elementPos);
+ }
+ return null;
+ });
+ break;
+ case NULL:
+ break;
+ case BOOLEAN:
+ getBoolean(value, pos);
+ break;
+ case LONG:
+ getLong(value, pos);
+ break;
+ case STRING:
+ getString(value, pos);
+ break;
+ case DOUBLE:
+ getDouble(value, pos);
+ break;
+ case DECIMAL:
+ getDecimal(value, pos);
+ break;
+ case DATE:
+ case TIMESTAMP:
+ case TIMESTAMP_NTZ:
+ getLong(value, pos);
+ break;
+ case FLOAT:
+ getFloat(value, pos);
+ break;
+ case BINARY:
+ getBinary(value, pos);
+ break;
+ case UUID:
+ getUuid(value, pos);
+ break;
+ default:
+ // This is practically unreachable because we handle all possible types. It only
+ // intends to ensure we don't forget adding a new case when adding a new type.
+ throw malformedVariant();
+ }
+ }
+
// Get a key at `id` in the variant metadata.
// Throw `MALFORMED_VARIANT` if the variant is malformed. An out-of-bound `id` is also considered
// a malformed variant because it is read from the corresponding variant value.
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst b/python/docs/source/reference/pyspark.sql/functions.rst
index 23666983fd9e..d6810722e79a 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -586,6 +586,7 @@ VARIANT Functions
.. autosummary::
:toctree: api/
+ is_valid_variant
is_variant_null
parse_json
schema_of_variant
diff --git a/python/pyspark/sql/connect/functions/builtin.py b/python/pyspark/sql/connect/functions/builtin.py
index d7c8a2284237..76e3f40a101a 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -2189,6 +2189,13 @@ def is_variant_null(v: "ColumnOrName") -> Column:
is_variant_null.__doc__ = pysparkfuncs.is_variant_null.__doc__
+def is_valid_variant(v: "ColumnOrName") -> Column:
+ return _invoke_function("is_valid_variant", _to_col(v))
+
+
+is_valid_variant.__doc__ = pysparkfuncs.is_valid_variant.__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 c1ecaeb8032f..27db280be86d 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -470,6 +470,7 @@ __all__ = [ # noqa: F405
"schema_of_json",
"to_json",
# VARIANT Functions
+ "is_valid_variant",
"is_variant_null",
"parse_json",
"schema_of_variant",
diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py
index cc8297297794..5e0fd250c8a9 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -21224,6 +21224,35 @@ def is_variant_null(v: "ColumnOrName") -> Column:
return _invoke_function("is_variant_null", _to_java_column(v))
+@_try_remote_functions
+def is_valid_variant(v: "ColumnOrName") -> Column:
+ """
+ Check if a variant value is valid. Returns true if the variant is valid, false if it is
+ malformed, and NULL if the input is NULL.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ v : :class:`~pyspark.sql.Column` or str
+ a variant column or column name
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ a boolean column indicating whether the variant value is valid
+
+ Examples
+ --------
+ >>> df = spark.createDataFrame([ {'json': '''{ "a" : 1 }'''} ])
+ >>> df.select(is_valid_variant(parse_json(df.json)).alias("r")).collect()
+ [Row(r=True)]
+ """
+ from pyspark.sql.classic.column import _to_java_column
+
+ return _invoke_function("is_valid_variant", _to_java_column(v))
+
+
@_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 65c7fc0c29ec..417e93760ef3 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -3431,6 +3431,7 @@ class FunctionsTestsMixin:
self.assertEqual([r[0] for r in resultDf.collect()], expected)
check(df.select(F.is_variant_null(v)), [False, False])
+ check(df.select(F.is_valid_variant(v)), [True, True])
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 ac0b45925194..ccc7820d5d95 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
@@ -9585,6 +9585,17 @@ object functions {
*/
def is_variant_null(v: Column): Column = Column.fn("is_variant_null", v)
+ /**
+ * Check if a variant value is valid. Returns true if the variant is valid, false if it is
+ * malformed, and NULL if the input is NULL.
+ *
+ * @param v
+ * a variant column.
+ * @group variant_funcs
+ * @since 4.2.0
+ */
+ def is_valid_variant(v: Column): Column = Column.fn("is_valid_variant", v)
+
/**
* 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 b77cdc15f845..34b9559c5851 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
@@ -979,6 +979,7 @@ object FunctionRegistry {
expression[SchemaOfVariant]("schema_of_variant"),
expression[SchemaOfVariantAgg]("schema_of_variant_agg"),
expression[ToVariantObject]("to_variant_object"),
+ expression[IsValidVariant]("is_valid_variant"),
// 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 61e51988c1e6..1f23ba05bc03 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
@@ -70,6 +70,9 @@ object VariantExpressionEvalUtils {
}
}
+ def isValidVariant(input: VariantVal): Boolean =
+ VariantUtil.isValidVariant(input.getValue, input.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 a7e19dfb1948..a928c59fdfc5 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
@@ -952,3 +952,35 @@ case class SchemaOfVariantAgg(
override protected def withNewChildInternal(newChild: Expression): Expression =
copy(child = newChild)
}
+
+@ExpressionDescription(
+ usage = "_FUNC_(v) - Returns true if the variant is valid, false if it is malformed, " +
+ "NULL if `v` is NULL.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_(parse_json('null'));
+ true
+ > SELECT _FUNC_(parse_json('[{"b":true,"a":0}]'));
+ true
+ """,
+ since = "4.2.0",
+ group = "variant_funcs"
+)
+case class IsValidVariant(child: Expression) extends UnaryExpression
+ with Predicate with ExpectsInputTypes with RuntimeReplaceable {
+
+ override lazy val replacement: Expression = StaticInvoke(
+ VariantExpressionEvalUtils.getClass,
+ BooleanType,
+ "isValidVariant",
+ Seq(child),
+ inputTypes,
+ returnNullable = false)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(VariantType)
+
+ override def prettyName: String = "is_valid_variant"
+
+ override protected def withNewChildInternal(newChild: Expression): IsValidVariant =
+ copy(child = newChild)
+}
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 ef2618e8455f..dc42cd7fbb2e 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
@@ -1053,4 +1053,151 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
}
}
}
+
+ test("is_valid_variant") {
+ val emptyMetadata = Array[Byte](VERSION, 0, 0)
+
+ // The row cannot be converted to string because the `VariantVal` may be malformed (toString
+ // will throw an exception).
+ class NoDisplayGenericInternalRow(values: Array[Any]) extends GenericInternalRow(values) {
+ override def toString: String = "NoDisplayGenericInternalRow"
+ }
+
+ def valid(value: Array[Byte], metadata: Array[Byte] = emptyMetadata): Unit = {
+ val row = new NoDisplayGenericInternalRow(Array(new VariantVal(value, metadata)))
+ val v = BoundReference(0, VariantType, nullable = true)
+ checkEvaluation(IsValidVariant(v), true, row)
+ }
+
+ def invalid(value: Array[Byte], metadata: Array[Byte] = emptyMetadata): Unit = {
+ val row = new NoDisplayGenericInternalRow(Array(new VariantVal(value, metadata)))
+ val v = BoundReference(0, VariantType, nullable = true)
+ checkEvaluation(IsValidVariant(v), false, row)
+ }
+
+ // Valid primitives.
+ valid(Array(primitiveHeader(NULL)))
+ valid(Array(primitiveHeader(TRUE)))
+ valid(Array(primitiveHeader(FALSE)))
+ valid(Array(primitiveHeader(INT1), 1))
+ valid(Array(primitiveHeader(INT2), 1, 0))
+ valid(Array(primitiveHeader(INT4), 1, 0, 0, 0))
+ valid(Array(primitiveHeader(INT8), 1, 0, 0, 0, 0, 0, 0, 0))
+ valid(Array(primitiveHeader(DOUBLE), 0, 0, 0, 0, 0, 0, 0, 0))
+ valid(Array(primitiveHeader(DECIMAL4), 0, 1, 0, 0, 0))
+ valid(Array(primitiveHeader(FLOAT), 0, 0, 0, 0))
+ valid(Array(primitiveHeader(DATE), 0, 0, 0, 0))
+ valid(Array(primitiveHeader(TIMESTAMP), 0, 0, 0, 0, 0, 0, 0, 0))
+ valid(Array(primitiveHeader(TIMESTAMP_NTZ), 0, 0, 0, 0, 0, 0, 0, 0))
+ valid(Array(shortStrHeader(3), 'a', 'b', 'c'))
+ valid(Array(primitiveHeader(LONG_STR), 2, 0, 0, 0, 'a', 'b'))
+ valid(Array(primitiveHeader(BINARY), 2, 0, 0, 0, 1, 2))
+ valid(Array(primitiveHeader(UUID)) ++ createArray[Byte](16, 0.toByte))
+
+ // Malformed primitives: truncated content.
+ invalid(Array(primitiveHeader(INT8), 0, 0, 0, 0, 0, 0, 0))
+ invalid(Array(primitiveHeader(DECIMAL4)))
+ invalid(Array(primitiveHeader(DECIMAL8)))
+ invalid(Array(primitiveHeader(DECIMAL16)))
+ invalid(Array(primitiveHeader(DECIMAL16)) ++ createArray[Byte](16, 0.toByte))
+ invalid(Array(shortStrHeader(2), 'x'))
+ invalid(Array(primitiveHeader(LONG_STR), 0, 0, 0))
+ invalid(Array(primitiveHeader(LONG_STR), 1, 0, 0, 0))
+
+ // Valid array.
+ valid(Array(arrayHeader(false, 1),
+ /* size */ 2,
+ /* offset list */ 0, 1, 2,
+ /* element data */ primitiveHeader(TRUE), primitiveHeader(FALSE)))
+
+ // Valid empty array.
+ valid(Array(arrayHeader(false, 1),
+ /* size */ 0,
+ /* offset list */ 0))
+
+ // Malformed array: size is 1 but no content.
+ invalid(Array(arrayHeader(false, 1),
+ /* size */ 1,
+ /* offset list */ 0))
+
+ // Malformed array: requires 4-byte size but only one byte given.
+ invalid(Array(arrayHeader(true, 1),
+ /* size */ 0,
+ /* offset list */ 0))
+
+ // Malformed array: offset out of bound.
+ invalid(Array(arrayHeader(false, 1),
+ /* size */ 1,
+ /* offset list */ 1, 1))
+
+ // Malformed array: nested element is malformed.
+ invalid(Array(arrayHeader(false, 1),
+ /* size */ 1,
+ /* offset list */ 0, 2,
+ /* element data: INT8 with only 1 byte */ primitiveHeader(INT8), 0))
+
+ // Valid object.
+ val metadata = Array[Byte](VERSION, 2, 0, 1, 2) ++ Array[Byte]('a', 'b')
+ valid(Array(objectHeader(false, 1, 1),
+ /* size */ 2,
+ /* id list */ 0, 1,
+ /* offset list */ 0, 2, 4,
+ /* field data */ primitiveHeader(INT1), 1, primitiveHeader(INT1), 2), metadata)
+
+ // Valid empty object.
+ valid(Array(objectHeader(false, 1, 1),
+ /* size */ 0,
+ /* offset list */ 0))
+
+ // Malformed object: id out of bound.
+ invalid(Array(objectHeader(false, 1, 1),
+ /* size */ 1,
+ /* id list */ 0,
+ /* offset list */ 0, 2,
+ /* field data */ primitiveHeader(INT1), 1))
+
+ // Malformed object: offset out of bound.
+ invalid(Array(objectHeader(false, 1, 1),
+ /* size */ 1,
+ /* id list */ 0,
+ /* offset list */ 5, 0,
+ /* field data */ primitiveHeader(INT1), 1), metadata)
+
+ // Malformed object: nested value is malformed.
+ invalid(Array(objectHeader(false, 1, 1),
+ /* size */ 1,
+ /* id list */ 0,
+ /* offset list */ 0, 2,
+ /* field data: INT8 with only 1 byte */ primitiveHeader(INT8), 0), metadata)
+
+ // Unknown primitive type (type info 17 is not defined).
+ invalid(Array(primitiveHeader(17)))
+
+ // Malformed metadata: version is not 1.
+ invalid(Array(primitiveHeader(INT1), 0), Array[Byte](3, 0, 0))
+ invalid(Array(primitiveHeader(INT1), 0), Array[Byte](2, 0, 0))
+
+ // Malformed metadata: offset > nextOffset for key id 0.
+ invalid(Array(objectHeader(false, 1, 1),
+ /* size */ 1,
+ /* id list */ 0,
+ /* offset list */ 0, 2,
+ /* field data */ primitiveHeader(INT1), 1),
+ Array[Byte](VERSION, 1, 2, 1) ++ Array[Byte]('a', 'b'))
+
+ // Malformed metadata: truncated offset list (declares dict size 1 but is missing nextOffset).
+ invalid(Array(objectHeader(false, 1, 1),
+ /* size */ 1,
+ /* id list */ 0,
+ /* offset list */ 0, 2,
+ /* field data */ primitiveHeader(INT1), 1),
+ Array[Byte](VERSION, 1, 0))
+
+ // Valid metadata formats: extra bits are ignored.
+ valid(Array(primitiveHeader(TRUE)), Array[Byte](VERSION | 1 << 4, 0, 0))
+ valid(Array(primitiveHeader(TRUE)), Array[Byte](VERSION | 1 << 5, 0, 0))
+
+ // Null input.
+ checkEvaluation(IsValidVariant(Literal.create(null, VariantType)), 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 87c01dc4b35c..16a2bf85de4a 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
@@ -2728,6 +2728,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
fn.is_variant_null(fn.parse_json(fn.col("g")))
}
+ functionTest("is_valid_variant") {
+ fn.is_valid_variant(fn.parse_json(fn.col("g")))
+ }
+
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_is_valid_variant.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_is_valid_variant.explain
new file mode 100644
index 000000000000..324e72dfbcc1
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_is_valid_variant.explain
@@ -0,0 +1,2 @@
+Project [static_invoke(VariantExpressionEvalUtils.isValidVariant(static_invoke(VariantExpressionEvalUtils.parseJson(g#0, false, true)))) AS is_valid_variant(parse_json(g))#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_is_valid_variant.json b/sql/connect/common/src/test/resources/query-tests/queries/function_is_valid_variant.json
new file mode 100644
index 000000000000..cdf4cca04fda
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/queries/function_is_valid_variant.json
@@ -0,0 +1,83 @@
+{
+ "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": "is_valid_variant",
+ "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"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "is_valid_variant",
+ "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_is_valid_variant.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/function_is_valid_variant.proto.bin
new file mode 100644
index 000000000000..b327ddbfc087
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/function_is_valid_variant.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 4a43bbcf853c..b195f7615687 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
@@ -542,6 +542,7 @@
| org.apache.spark.sql.catalyst.expressions.st.ST_GeomFromWKB | st_geomfromwkb | SELECT hex(st_asbinary(st_geomfromwkb(X'0101000000000000000000F03F0000000000000040'))) | struct<hex(st_asbinary(st_geomfromwkb(X'0101000000000000000000F03F0000000000000040', 0))):string> |
| org.apache.spark.sql.catalyst.expressions.st.ST_SetSrid | st_setsrid | SELECT st_srid(st_setsrid(ST_GeogFromWKB(X'0101000000000000000000F03F0000000000000040'), 4326)) | struct<st_srid(st_setsrid(st_geogfromwkb(X'0101000000000000000000F03F0000000000000040'), 4326)):int> |
| org.apache.spark.sql.catalyst.expressions.st.ST_Srid | st_srid | SELECT st_srid(st_geogfromwkb(X'0101000000000000000000F03F0000000000000040')) | struct<st_srid(st_geogfromwkb(X'0101000000000000000000F03F0000000000000040')):int> |
+| org.apache.spark.sql.catalyst.expressions.variant.IsValidVariant | is_valid_variant | SELECT is_valid_variant(parse_json('null')) | struct<is_valid_variant(parse_json(null)):boolean> |
| org.apache.spark.sql.catalyst.expressions.variant.IsVariantNull | is_variant_null | SELECT is_variant_null(parse_json('null')) | struct<is_variant_null(parse_json(null)):boolean> |
| org.apache.spark.sql.catalyst.expressions.variant.ParseJsonExpressionBuilder | parse_json | SELECT parse_json('{"a":1,"b":0.8}') | struct<parse_json({"a":1,"b":0.8}):variant> |
| org.apache.spark.sql.catalyst.expressions.variant.SchemaOfVariant | schema_of_variant | SELECT schema_of_variant(parse_json('null')) | struct<schema_of_variant(parse_json(null)):string> |
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org