[SPARK-57927][SQL] Add json_valid function for JSON syntax validation

python/docs/source/reference/pyspark.sql/functions.rst

@@ -586,6 +586,7 @@ JSON Functions
     get_json_object
     json_array_length
     json_object_keys
+    json_valid
     json_tuple
     schema_of_json
     to_json

python/pyspark/sql/connect/functions/builtin.py

@@ -2060,6 +2060,13 @@ def json_object_keys(col: "ColumnOrName") -> Column:
 json_object_keys.__doc__ = pysparkfuncs.json_object_keys.__doc__
 
 
+def json_valid(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("json_valid", col)
+
+
+json_valid.__doc__ = pysparkfuncs.json_valid.__doc__
+
+
 def inline(col: "ColumnOrName") -> Column:
     return _invoke_function_over_columns("inline", col)
 

python/pyspark/sql/functions/__init__.py

@@ -470,6 +470,7 @@
     "get_json_object",
     "json_array_length",
     "json_object_keys",
+    "json_valid",
     "json_tuple",
     "schema_of_json",
     "to_json",

python/pyspark/sql/functions/builtin.py

@@ -22168,6 +22168,36 @@ def json_object_keys(col: "ColumnOrName") -> Column:
     return _invoke_function_over_columns("json_object_keys", col)
 
 
+@_try_remote_functions
+def json_valid(col: "ColumnOrName") -> Column:
+    """
+    Returns true if the input is a valid JSON string, false otherwise. Returns null if the
+    input is null. Parsing follows the same lenient rules as the other JSON functions
+    (e.g. :func:`get_json_object`), so single-quoted strings and unescaped control characters
+    are accepted.
+
+    .. versionadded:: 4.3.0
+
+    Parameters
+    ----------
+    col: :class:`~pyspark.sql.Column` or str
+        target column to compute on.
+
+    Returns
+    -------
+    :class:`~pyspark.sql.Column`
+        a boolean indicating whether the input is a valid JSON string.
+
+    Examples
+    --------
+    >>> df = spark.createDataFrame(
+    ...     [(None,), ('{"a": 1}',), ('[1, 2, 3]',), ('invalid',), ('{"a":1} x',)], ['data'])
+    >>> df.select(json_valid(df.data).alias('r')).collect()
+    [Row(r=None), Row(r=True), Row(r=True), Row(r=False), Row(r=False)]
+    """
+    return _invoke_function_over_columns("json_valid", col)
+
+
 # TODO: Fix and add an example for StructType with Spark Connect
 #   e.g., StructType([StructField("a", IntegerType())])
 @_try_remote_functions

sql/api/src/main/scala/org/apache/spark/sql/functions.scala

@@ -10011,6 +10011,16 @@ object functions {
    */
   def json_object_keys(e: Column): Column = Column.fn("json_object_keys", e)
 
+  /**
+   * Returns true if the input is a valid JSON string, false otherwise. Returns null if the input
+   * is null. Parsing follows the same lenient rules as the other JSON functions (e.g.
+   * `get_json_object`), so single-quoted strings and unescaped control characters are accepted.
+   *
+   * @group json_funcs
+   * @since 4.3.0
+   */
+  def json_valid(e: Column): Column = Column.fn("json_valid", e)
+
   // scalastyle:off line.size.limit
   /**
    * (Scala-specific) Converts a column containing a `StructType`, `ArrayType` or a `MapType` into

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionUtils.java

@@ -54,6 +54,23 @@ public static Integer lengthOfJsonArray(UTF8String json) {
     }
   }
 
+  public static Boolean isJsonValid(UTF8String json) {
+    try (JsonParser jsonParser =
+        CreateJacksonParser.utf8String(SharedFactory.jsonFactory(), json)) {
+      // An empty or whitespace-only input is not valid JSON.
+      if (jsonParser.nextToken() == null) {
+        return false;
+      }
+      // Consume the whole first value (including all nested children).
+      jsonParser.skipChildren();
+      // The input is valid only if the first value is also the last, i.e. there is no
+      // trailing content after the root JSON value.
+      return jsonParser.nextToken() == null;
+    } catch (IOException e) {
+      return false;
+    }
+  }
+
   public static GenericArrayData jsonObjectKeys(UTF8String json) {
     try (JsonParser jsonParser =
         CreateJacksonParser.utf8String(SharedFactory.jsonFactory(), json)) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -974,6 +974,7 @@ object FunctionRegistry {
     expression[SchemaOfJson]("schema_of_json"),
     expression[LengthOfJsonArray]("json_array_length"),
     expression[JsonObjectKeys]("json_object_keys"),
+    expression[JsonValid]("json_valid"),
 
     // Variant
     expressionBuilder("parse_json", ParseJsonExpressionBuilder),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

@@ -693,3 +693,54 @@ case class JsonObjectKeys(child: Expression)
   override protected def withNewChildInternal(newChild: Expression): JsonObjectKeys =
     copy(child = newChild)
 }
+
+/**
+ * A function that returns true if the input is a valid JSON string, false otherwise.
+ */
+@ExpressionDescription(
+  usage = "_FUNC_(jsonString) - Returns true if `jsonString` is a valid JSON string, " +
+    "false otherwise. Returns null if the input is null.",
+  arguments = """
+    Arguments:
+      * jsonString - A string to be validated as JSON. Parsing follows the same lenient rules as
+          the other JSON functions (e.g. `get_json_object`), so single-quoted strings and
+          unescaped control characters are accepted.
+  """,
+  examples = """
+    Examples:
+      > SELECT _FUNC_('{"a":1}');
+        true
+      > SELECT _FUNC_('[1, 2, 3]');
+        true
+      > SELECT _FUNC_('invalid');
+        false
+      > SELECT _FUNC_('{"a":1} garbage');
+        false
+      > SELECT _FUNC_('');
+        false
+  """,
+  group = "json_funcs",
+  since = "4.3.0"
+)
+case class JsonValid(child: Expression)
+  extends UnaryExpression
+  with ExpectsInputTypes
+  with RuntimeReplaceable {
+
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(StringTypeWithCollation(supportsTrimCollation = true))
+  override def dataType: DataType = BooleanType
+  override def nullable: Boolean = true
+  override def prettyName: String = "json_valid"
+
+  override def replacement: Expression = StaticInvoke(
+    classOf[JsonExpressionUtils],
+    dataType,
+    "isJsonValid",
+    Seq(child),
+    inputTypes
+  )
+
+  override protected def withNewChildInternal(newChild: Expression): JsonValid =
+    copy(child = newChild)
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala

@@ -909,6 +909,36 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     }
   }
 
+  test("json_valid") {
+    Seq(
+      // null input returns null
+      (null, null),
+      // valid JSON values
+      ("""{"a":1}""", true),
+      ("""{"a": "b", "c": [1, 2, 3]}""", true),
+      ("[1, 2, 3]", true),
+      ("[]", true),
+      ("{}", true),
+      ("\"a string\"", true),
+      ("123", true),
+      ("true", true),
+      ("null", true),
+      // Lenient parsing (shared with the other JSON functions): single quotes are accepted.
+      ("{'a':1}", true),
+      // invalid JSON values
+      ("", false),
+      ("   ", false),
+      ("invalid", false),
+      ("""{"a":1} garbage""", false),
+      ("[1, 2, 3", false),
+      ("""{"a": }""", false)
+    ).foreach {
+      case (input, expected) =>
+        val literal = if (input == null) Literal.create(null, StringType) else Literal(input)
+        checkEvaluation(JsonValid(literal), expected)
+    }
+  }
+
   test("SPARK-35320: from_json should fail with a key type different of StringType") {
     Seq(
       (MapType(IntegerType, StringType), """{"1": "test"}"""),

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala

@@ -3101,6 +3101,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
     fn.json_object_keys(fn.col("g"))
   }
 
+  functionTest("json_valid") {
+    fn.json_valid(fn.col("g"))
+  }
+
   functionTest("mask with specific upperChar lowerChar digitChar otherChar") {
     fn.mask(fn.col("g"), fn.lit('X'), fn.lit('x'), fn.lit('n'), fn.lit('*'))
   }

sql/connect/common/src/test/resources/query-tests/explain-results/function_json_valid.explain

@@ -0,0 +1,2 @@
+Project [static_invoke(JsonExpressionUtils.isJsonValid(g#0)) AS json_valid(g)#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

sql/connect/common/src/test/resources/query-tests/queries/function_json_valid.json

@@ -0,0 +1,60 @@
+{
+  "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": "json_valid",
+        "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": "json_valid",
+              "fileName": "functions.scala"
+            }, {
+              "classLoaderName": "app",
+              "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+              "methodName": "~~trimmed~anonfun~~",
+              "fileName": "PlanGenerationTestSuite.scala"
+            }]
+          }
+        }
+      }
+    }]
+  }
+}

sql/core/src/test/resources/sql-functions/sql-expression-schema.md

@@ -186,6 +186,7 @@
 | org.apache.spark.sql.catalyst.expressions.JsonObjectKeys | json_object_keys | SELECT json_object_keys('{}') | struct<json_object_keys({}):array<string>> |
 | org.apache.spark.sql.catalyst.expressions.JsonToStructs | from_json | SELECT from_json('{"a":1, "b":0.8}', 'a INT, b DOUBLE') | struct<from_json({"a":1, "b":0.8}):struct<a:int,b:double>> |
 | org.apache.spark.sql.catalyst.expressions.JsonTuple | json_tuple | SELECT json_tuple('{"a":1, "b":2}', 'a', 'b') | struct<c0:string,c1:string> |
+| org.apache.spark.sql.catalyst.expressions.JsonValid | json_valid | SELECT json_valid('{"a":1}') | struct<json_valid({"a":1}):boolean> |
 | org.apache.spark.sql.catalyst.expressions.KllSketchGetNBigint | kll_sketch_get_n_bigint | SELECT kll_sketch_get_n_bigint(kll_sketch_agg_bigint(col)) FROM VALUES (1), (2), (3), (4), (5) tab(col) | struct<kll_sketch_get_n_bigint(kll_sketch_agg_bigint(col)):bigint> |
 | org.apache.spark.sql.catalyst.expressions.KllSketchGetNDouble | kll_sketch_get_n_double | SELECT kll_sketch_get_n_double(kll_sketch_agg_double(col)) FROM VALUES (CAST(1.0 AS DOUBLE)), (CAST(2.0 AS DOUBLE)), (CAST(3.0 AS DOUBLE)), (CAST(4.0 AS DOUBLE)), (CAST(5.0 AS DOUBLE)) tab(col) | struct<kll_sketch_get_n_double(kll_sketch_agg_double(col)):bigint> |
 | org.apache.spark.sql.catalyst.expressions.KllSketchGetNFloat | kll_sketch_get_n_float | SELECT kll_sketch_get_n_float(kll_sketch_agg_float(col)) FROM VALUES (CAST(1.0 AS FLOAT)), (CAST(2.0 AS FLOAT)), (CAST(3.0 AS FLOAT)), (CAST(4.0 AS FLOAT)), (CAST(5.0 AS FLOAT)) tab(col) | struct<kll_sketch_get_n_float(kll_sketch_agg_float(col)):bigint> |