dataflow-expression: Add json_valid MySQL function

Determines if text is valid JSON. It has null propagation but only for
text types. Non-text types return false.

This commit adds `DfValue::is_any_json_like` which returns `true` for
types that can be treated like JSON/JSONB.

Fixes: ENG-1549
Release-Note-Core: Added support for the MySQL function `json_valid`.
Change-Id: I5347c9571aa96f4975fca72597bc24096a16979c
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/3996
Tested-by: Buildkite CI
Reviewed-by: Griffin Smith <griffin@readyset.io>

dataflow-expression/src/eval/builtins.rs

@@ -565,6 +565,21 @@ impl BuiltinFunction {
                     }
                 }
             }
+            BuiltinFunction::JsonValid(expr) => {
+                let value = expr.eval(record)?;
+
+                let valid = if expr.ty().is_known() && !expr.ty().is_any_json_like() {
+                    // Known non-json-like types return `false` and don't null-propagate.
+                    false
+                } else {
+                    // Non-text unknown-type values return `false`.
+                    <&str>::try_from(non_null!(&value))
+                        .map(|json| serde_json::from_str::<serde::de::IgnoredAny>(json).is_ok())
+                        .unwrap_or_default()
+                };
+
+                Ok(valid.into())
+            }
             BuiltinFunction::JsonTypeof(expr) => {
                 let json = non_null!(expr.eval(record)?).to_json()?;
                 Ok(get_json_value_type(&json).into())
@@ -1793,6 +1808,43 @@ mod tests {
         use super::*;
         use crate::utils::normalize_json;
 
+        #[test]
+        fn json_valid() {
+            #[track_caller]
+            fn test(json_expr: &str, expected: Option<bool>) {
+                let expr = format!("json_valid({json_expr})");
+
+                assert_eq!(
+                    eval_expr(&expr, MySQL),
+                    expected.into(),
+                    "incorrect result for for `{expr}`"
+                );
+            }
+
+            test("null", None);
+
+            test("'null'", Some(true));
+            test("'1'", Some(true));
+            test("'1.5'", Some(true));
+            test(r#"'"hi"'"#, Some(true));
+            test("'[]'", Some(true));
+            test("'[42]'", Some(true));
+            test("'{}'", Some(true));
+            test(r#"'{ "a": 42 }'"#, Some(true));
+
+            test("''", Some(false));
+            test("'hello'", Some(false));
+            test("'['", Some(false));
+            test("'{'", Some(false));
+            test("'+'", Some(false));
+            test("'-'", Some(false));
+
+            // Non-text types are allowed and return false.
+            test("1", Some(false));
+            test("1.5", Some(false));
+            test("dayofweek(null)", Some(false));
+        }
+
         #[test]
         fn json_array_length() {
             #[track_caller]

dataflow-expression/src/lib.rs

@@ -48,6 +48,8 @@ pub enum BuiltinFunction {
     Round(Expr, Expr),
     /// [`json_depth`](https://dev.mysql.com/doc/refman/8.0/en/json-attribute-functions.html#function_json-depth)
     JsonDepth(Expr),
+    /// [`json_valid`](https://dev.mysql.com/doc/refman/8.0/en/json-attribute-functions.html#function_json-valid)
+    JsonValid(Expr),
     /// [`json[b]_typeof`](https://www.postgresql.org/docs/current/functions-json.html)
     JsonTypeof(Expr),
     /// [`json[b]_array_length`](https://www.postgresql.org/docs/current/functions-json.html)
@@ -112,6 +114,7 @@ impl BuiltinFunction {
             DateFormat { .. } => "date_format",
             Round { .. } => "round",
             JsonDepth { .. } => "json_depth",
+            JsonValid { .. } => "json_valid",
             JsonTypeof { .. } => "json_typeof",
             JsonArrayLength { .. } => "json_array_length",
             JsonStripNulls { .. } => "json_strip_nulls",
@@ -163,8 +166,8 @@ impl Display for BuiltinFunction {
             Round(arg1, precision) => {
                 write!(f, "({}, {})", arg1, precision)
             }
-            JsonDepth(arg) | JsonTypeof(arg) | JsonArrayLength(arg) | JsonStripNulls(arg)
-            | JsonbPretty(arg) => {
+            JsonDepth(arg) | JsonValid(arg) | JsonTypeof(arg) | JsonArrayLength(arg)
+            | JsonStripNulls(arg) | JsonbPretty(arg) => {
                 write!(f, "({})", arg)
             }
             JsonExtractPath { json, keys } => {

dataflow-expression/src/lower.rs

@@ -214,6 +214,7 @@ impl BuiltinFunction {
                 (Self::Round(expr, prec), ty)
             }
             "json_depth" => (Self::JsonDepth(next_arg()?), DfType::Int),
+            "json_valid" => (Self::JsonValid(next_arg()?), DfType::BigInt),
             "json_typeof" | "jsonb_typeof" => (
                 Self::JsonTypeof(next_arg()?),
                 // Always returns text containing the JSON type.

readyset-data/src/type.rs

@@ -431,6 +431,12 @@ impl DfType {
         matches!(self, Self::Json | Self::Jsonb)
     }
 
+    /// Returns `true` if this is any JSON-like type.
+    #[inline]
+    pub fn is_any_json_like(&self) -> bool {
+        self.is_any_json() || self.is_any_text()
+    }
+
     /// Returns `true` if this is the boolean type.
     #[inline]
     pub fn is_bool(&self) -> bool {