From cf0a4de671f7c42b03b318334d593d0849d5bfd4 Mon Sep 17 00:00:00 2001 From: Eric Wei Date: Fri, 12 Jun 2026 14:50:04 -0700 Subject: [PATCH] [analytics-engine] Add json_valid Rust UDF to the DataFusion backend Wires PPL json_valid through the analytics-engine route (PPL -> Calcite -> Substrait -> DataFusion): ScalarFunction.JSON_VALID enum, JsonValidAdapter (rewrites Calcite IS_JSON_VALUE -> local json_valid op), project + filter op registration, substrait FunctionMapping, yaml signature, and the Rust UDF. json_valid is registered as a FILTER op (not just PROJECT) so it works as a WHERE predicate (e.g. `where json_valid(col)` / `where not json_valid(col)`), same shape as cidrmatch. Semantics match the legacy SQL-plugin JsonUtils.isValidJson (Jackson readTree): malformed -> false, NULL -> NULL, and empty/whitespace -> true (Jackson returns MissingNode without throwing; serde rejects empty input, so is_valid_json special-cases it to preserve parity with the JsonFunctionsIT fixture). Complements #22130 (json/json_object/json_array); json_valid is independent. Rust unit tests 10/10, Java adapter tests 9/9. Signed-off-by: Eric Wei --- .../analytics/spi/ScalarFunction.java | 12 + .../analytics/spi/ScalarFunctionTests.java | 12 + .../rust/src/udf/json_valid.rs | 273 ++++++++++++++++++ .../rust/src/udf/mod.rs | 2 + .../DataFusionAnalyticsBackendPlugin.java | 8 +- .../DataFusionFragmentConvertor.java | 1 + .../be/datafusion/JsonFunctionAdapters.java | 34 +++ .../opensearch_scalar_functions.yaml | 6 + .../datafusion/JsonFunctionAdaptersTests.java | 68 +++++ 9 files changed, 415 insertions(+), 1 deletion(-) create mode 100644 sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_valid.rs diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java index 2c21a42592aa0..a4435ff9b979d 100644 --- a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java +++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java @@ -355,6 +355,18 @@ public enum ScalarFunction { JSON_EXTRACT_ALL(Category.SCALAR, SqlKind.OTHER_FUNCTION), JSON_KEYS(Category.SCALAR, SqlKind.OTHER_FUNCTION), JSON_SET(Category.SCALAR, SqlKind.OTHER_FUNCTION), + /** + * PPL {@code json_valid(str)} — resolves through the SQL plugin's + * {@code PPLFuncImpTable} to {@link SqlStdOperatorTable#IS_JSON_VALUE}, which is + * a {@link org.apache.calcite.sql.SqlPostfixOperator} named {@code "IS JSON VALUE"} + * with {@link SqlKind#OTHER}. Neither name-based {@link #valueOf(String)} nor + * {@link SqlKind}-based resolution matches. The {@link SqlKind#OTHER_FUNCTION} + * declaration opts out of the {@link #fromSqlKind(SqlKind)} scan (which would + * otherwise break {@code testFromSqlKindReturnsNullForOtherKind} by claiming + * {@code SqlKind.OTHER}); resolution happens via the {@code referenceOperator} + * singleton-identity match — same pattern as {@link #CONCAT}. + */ + JSON_VALID(Category.SCALAR, SqlKind.OTHER_FUNCTION, SqlStdOperatorTable.IS_JSON_VALUE), PATTERN_PARSER(Category.SCALAR, SqlKind.OTHER_FUNCTION), diff --git a/sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ScalarFunctionTests.java b/sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ScalarFunctionTests.java index 3fcc68efdcd59..652ce3b02060f 100644 --- a/sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ScalarFunctionTests.java +++ b/sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ScalarFunctionTests.java @@ -98,6 +98,18 @@ public void testFromSqlOperatorResolvesVariadicConcatViaReferenceOperator() { assertSame(ScalarFunction.CONCAT_FUNCTION, ScalarFunction.fromSqlOperatorWithFallback(SqlLibraryOperators.CONCAT_FUNCTION)); } + public void testFromSqlOperatorResolvesJsonValidViaReferenceOperator() { + // PPL json_valid reaches AE as SqlStdOperatorTable.IS_JSON_VALUE — a SqlPostfixOperator + // named "IS JSON VALUE" with SqlKind.OTHER. Neither fromSqlKind (OTHER is unmapped) nor + // identifier-name valueOf ("IS JSON VALUE" != "JSON_VALID") resolves it; only the + // referenceOperator identity pin does. Pins the exact production resolution path so a + // future refactor can't silently regress json_valid to "No backend supports scalar + // function" at the AE route. + assertEquals("IS JSON VALUE", SqlStdOperatorTable.IS_JSON_VALUE.getName()); + assertEquals(SqlKind.OTHER, SqlStdOperatorTable.IS_JSON_VALUE.getKind()); + assertSame(ScalarFunction.JSON_VALID, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.IS_JSON_VALUE)); + } + // ── fromSqlOperatorWithFallback: identifier-name branch ──────────────────────────────── public void testFromSqlOperatorResolvesViaIdentifierName() { diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_valid.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_valid.rs new file mode 100644 index 0000000000000..713645fffcd8f --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/json_valid.rs @@ -0,0 +1,273 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +//! `json_valid(value)` — TRUE iff the input parses as JSON. +//! +//! Parity target: the legacy SQL-plugin `JsonUtils.isValidJson` (Jackson +//! `ObjectMapper.readTree`), which is the runtime behind the PPL `json_valid` +//! the `JsonFunctionsIT` suite asserts against. Contract: +//! * valid JSON → TRUE +//! * malformed input → FALSE +//! * NULL / missing input → FALSE (legacy returns `LITERAL_FALSE`, NOT null), +//! so `where not json_valid(col)` includes NULL rows +//! * empty / whitespace-only input → TRUE (Jackson `readTree("")` returns a +//! `MissingNode` without throwing; serde rejects it, so `is_valid_json` +//! special-cases it — see that fn) + +use std::sync::Arc; + +use datafusion::arrow::array::{ArrayRef, BooleanBuilder}; +use datafusion::arrow::datatypes::DataType; +use datafusion::common::{plan_err, ScalarValue}; +use datafusion::error::Result; +use datafusion::execution::context::SessionContext; +use datafusion::logical_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, +}; +use serde_json::Value; + +use super::json_common::StringArrayView; +use super::{coerce_args, CoerceMode}; + +pub fn register_all(ctx: &SessionContext) { + ctx.register_udf(ScalarUDF::from(JsonValidUdf::new())); +} + +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct JsonValidUdf { + signature: Signature, +} + +impl JsonValidUdf { + pub fn new() -> Self { + Self { + signature: Signature::user_defined(Volatility::Immutable), + } + } +} + +impl Default for JsonValidUdf { + fn default() -> Self { + Self::new() + } +} + +impl ScalarUDFImpl for JsonValidUdf { + fn name(&self) -> &str { + "json_valid" + } + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, arg_types: &[DataType]) -> Result { + if arg_types.len() != 1 { + return plan_err!("json_valid expects 1 argument, got {}", arg_types.len()); + } + Ok(DataType::Boolean) + } + + fn coerce_types(&self, arg_types: &[DataType]) -> Result> { + coerce_args("json_valid", arg_types, &[CoerceMode::Utf8]) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + if args.args.len() != 1 { + return plan_err!("json_valid expects 1 argument, got {}", args.args.len()); + } + let n = args.number_rows; + + if let ColumnarValue::Scalar(sv) = &args.args[0] { + // NULL input → FALSE (not NULL), matching the legacy SQL-plugin + // JsonUtils.isValidJson: `if (isNull || isMissing) return LITERAL_FALSE`. + let valid = match sv { + ScalarValue::Utf8(opt) + | ScalarValue::LargeUtf8(opt) + | ScalarValue::Utf8View(opt) => opt.as_deref().map(is_valid_json).unwrap_or(false), + _ => false, + }; + return Ok(ColumnarValue::Scalar(ScalarValue::Boolean(Some(valid)))); + } + + let arr = args.args[0].clone().into_array(n)?; + // CoerceMode::Utf8 preserves the observed string type, so the input may be + // Utf8 / LargeUtf8 / Utf8View — StringArrayView reads all three uniformly. + let strings = StringArrayView::from_array(&arr)?; + + let mut builder = BooleanBuilder::with_capacity(n); + for i in 0..n { + // NULL row → FALSE (legacy JsonUtils.isValidJson returns FALSE for null/missing), + // so `where not json_valid(col)` includes NULL rows as the IT expects. + builder.append_value(strings.cell(i).map(is_valid_json).unwrap_or(false)); + } + Ok(ColumnarValue::Array(Arc::new(builder.finish()) as ArrayRef)) + } +} + +/// TRUE iff `s` is a well-formed JSON document (object, array, or scalar). +/// Matches the legacy PPL contract in SQL-plugin `JsonUtils.isValidJson`, which uses Jackson +/// `ObjectMapper.readTree`. Jackson returns a `MissingNode` (no exception) for empty or +/// whitespace-only input, so the legacy function — and the `JsonFunctionsIT.test_json_valid` +/// fixture — treat `""` as VALID. serde_json's `from_str` instead errors on empty input, so we +/// special-case empty/whitespace to preserve parity; all other inputs use serde's RFC 8259 parse. +/// Callers handle the NULL-input case before dispatching here; this helper is total over `&str`. +fn is_valid_json(s: &str) -> bool { + // Jackson readTree("") / readTree(" ") → MissingNode (valid); serde would reject. Match legacy. + if s.trim().is_empty() { + return true; + } + serde_json::from_str::(s).is_ok() +} + +// ─── tests ────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use datafusion::arrow::array::{Array, BooleanArray, StringArray}; + use datafusion::arrow::datatypes::Field; + + #[test] + fn parses_every_json_shape_as_valid() { + // Legacy CalcitePPLJsonBuiltinFunctionIT.testJsonValid pins array input; + // objects / nested / scalars / whitespace-padded are covered here so future + // callers do not discover Jackson parity edges empirically. + assert!(is_valid_json("[1,2,3,4]")); + assert!(is_valid_json("{\"k\":1}")); + assert!(is_valid_json("{\"a\":[1,{\"b\":null}]}")); + assert!(is_valid_json("42")); + assert!(is_valid_json("\"s\"")); + assert!(is_valid_json("null")); + assert!(is_valid_json("true")); + assert!(is_valid_json(" [1, 2] ")); + } + + #[test] + fn rejects_malformed() { + // Pinned against legacy IT's {"invalid": "json"} case + garbage. + assert!(!is_valid_json("{\"invalid\": \"json\"")); + assert!(!is_valid_json("not-json")); + assert!(!is_valid_json("[1,2")); + } + + #[test] + fn empty_and_whitespace_are_valid() { + // Legacy JsonUtils.isValidJson uses Jackson readTree, which returns MissingNode (no throw) + // for empty / whitespace-only input — so the PPL contract (and JsonFunctionsIT's + // "json empty string" fixture row) treats these as VALID. serde would reject them, hence the + // special-case in is_valid_json. + assert!(is_valid_json("")); + assert!(is_valid_json(" ")); + } + + #[test] + fn return_type_is_boolean() { + let udf = JsonValidUdf::new(); + assert_eq!( + udf.return_type(&[DataType::Utf8]).unwrap(), + DataType::Boolean + ); + } + + #[test] + fn coerce_types_accepts_string_variants() { + let udf = JsonValidUdf::new(); + // CoerceMode::Utf8 preserves the observed string type (Utf8/LargeUtf8/Utf8View); + // the StringArrayView reader handles all three at execution time. + for t in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] { + assert_eq!( + udf.coerce_types(std::slice::from_ref(&t)).unwrap(), + vec![t.clone()] + ); + } + } + + #[test] + fn coerce_types_rejects_non_string() { + let udf = JsonValidUdf::new(); + let err = udf.coerce_types(&[DataType::Int64]).unwrap_err(); + assert!(err.to_string().contains("expected string")); + } + + #[test] + fn coerce_types_rejects_wrong_arity() { + let udf = JsonValidUdf::new(); + assert!(udf.coerce_types(&[]).is_err()); + assert!(udf.coerce_types(&[DataType::Utf8, DataType::Utf8]).is_err()); + } + + #[test] + fn invoke_column_null_input_is_false() { + // NULL/missing input → FALSE (not NULL), matching legacy JsonUtils.isValidJson + // (`if (isNull || isMissing) return LITERAL_FALSE`). This is what makes + // `where not json_valid(col)` include NULL rows, per JsonFunctionsIT.test_not_json_valid. + let udf = JsonValidUdf::new(); + let input = StringArray::from(vec![ + Some("[1,2,3,4]"), + None, + Some("{\"invalid\": \"json\""), + Some("42"), + Some(""), + ]); + let args = ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::new(input))], + number_rows: 5, + arg_fields: vec![], + return_field: Arc::new(Field::new("out", DataType::Boolean, true)), + config_options: Arc::new(datafusion::config::ConfigOptions::new()), + }; + let out = udf.invoke_with_args(args).unwrap(); + let arr = match out { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + let arr = arr.as_any().downcast_ref::().unwrap(); + assert!(!arr.is_null(1), "NULL input must produce FALSE, not NULL"); + assert!(arr.value(0)); + assert!(!arr.value(1), "NULL input → FALSE (legacy JsonUtils.isValidJson)"); + assert!(!arr.value(2)); + assert!(arr.value(3)); + // Empty string is VALID per the legacy Jackson contract (see empty_and_whitespace_are_valid). + assert!(arr.value(4)); + } + + #[test] + fn invoke_scalar_input_produces_scalar_output() { + let udf = JsonValidUdf::new(); + let args = ScalarFunctionArgs { + args: vec![ColumnarValue::Scalar(ScalarValue::Utf8(Some( + "[1,2,3,4]".into(), + )))], + number_rows: 1, + arg_fields: vec![], + return_field: Arc::new(Field::new("out", DataType::Boolean, true)), + config_options: Arc::new(datafusion::config::ConfigOptions::new()), + }; + match udf.invoke_with_args(args).unwrap() { + ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))) => {} + other => panic!("expected Boolean(Some(true)), got {other:?}"), + } + } + + #[test] + fn invoke_scalar_null_is_false() { + // NULL scalar → Boolean(Some(false)), matching legacy JsonUtils.isValidJson. + let udf = JsonValidUdf::new(); + let args = ScalarFunctionArgs { + args: vec![ColumnarValue::Scalar(ScalarValue::Utf8(None))], + number_rows: 1, + arg_fields: vec![], + return_field: Arc::new(Field::new("out", DataType::Boolean, true)), + config_options: Arc::new(datafusion::config::ConfigOptions::new()), + }; + match udf.invoke_with_args(args).unwrap() { + ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => {} + other => panic!("expected Boolean(Some(false)), got {other:?}"), + } + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs index 393bbf50c5096..2d99e173748e7 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mod.rs @@ -140,6 +140,7 @@ pub mod json_extract; pub mod json_extract_all; pub mod json_keys; pub mod json_set; +pub mod json_valid; pub mod makedate; pub mod maketime; pub mod minspan_bucket; @@ -189,6 +190,7 @@ pub fn register_all(ctx: &SessionContext) { json_extract_all::register_all(ctx); json_keys::register_all(ctx); json_set::register_all(ctx); + json_valid::register_all(ctx); makedate::register_all(ctx); maketime::register_all(ctx); minspan_bucket::register_all(ctx); diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java index 29381c4bc73f9..8ecc6100a0e44 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java @@ -116,7 +116,11 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP ScalarFunction.MOD, ScalarFunction.EARLIEST, ScalarFunction.LATEST, - ScalarFunction.CIDRMATCH + ScalarFunction.CIDRMATCH, + // json_valid returns BOOLEAN, so it is a valid filter predicate (e.g. `where + // json_valid(col)` / `where not json_valid(col)`). DataFusion evaluates the json_valid Rust + // UDF natively; same shape as CIDRMATCH. + ScalarFunction.JSON_VALID ); // Project-side scalar functions DataFusion can evaluate natively. Each entry corresponds to a @@ -330,6 +334,7 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP // FieldType.MAP rather than SUPPORTED_FIELD_TYPES, mirroring the ARRAY-return split). ScalarFunction.JSON_KEYS, ScalarFunction.JSON_SET, + ScalarFunction.JSON_VALID, // Array functions whose RETURN type is element-typed (not ARRAY itself), so the // capability lookup at OpenSearchProjectRule resolves the call's return type to a // standard scalar FieldType and matches against SUPPORTED_FIELD_TYPES. @@ -739,6 +744,7 @@ public Map scalarFunctionAdapters() { Map.entry(ScalarFunction.JSON_EXTRACT_ALL, new JsonFunctionAdapters.JsonExtractAllAdapter()), Map.entry(ScalarFunction.JSON_KEYS, new JsonFunctionAdapters.JsonKeysAdapter()), Map.entry(ScalarFunction.JSON_SET, new JsonFunctionAdapters.JsonSetAdapter()), + Map.entry(ScalarFunction.JSON_VALID, new JsonFunctionAdapters.JsonValidAdapter()), Map.entry(ScalarFunction.LATEST, new EarliestLatestAdapter.LatestAdapter()), Map.entry(ScalarFunction.PATTERN_PARSER, new PatternParserAdapter()), Map.entry(ScalarFunction.LIKE, new LikeAdapter()), diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java index 35e2e5246f7e0..66ab8230d9df4 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java @@ -193,6 +193,7 @@ public class DataFusionFragmentConvertor implements FragmentConvertor { FunctionMappings.s(JsonFunctionAdapters.JsonExtractAllAdapter.LOCAL_JSON_EXTRACT_ALL_OP, "json_extract_all"), FunctionMappings.s(JsonFunctionAdapters.JsonKeysAdapter.LOCAL_JSON_KEYS_OP, "json_keys"), FunctionMappings.s(JsonFunctionAdapters.JsonSetAdapter.LOCAL_JSON_SET_OP, "json_set"), + FunctionMappings.s(JsonFunctionAdapters.JsonValidAdapter.LOCAL_JSON_VALID_OP, "json_valid"), FunctionMappings.s(SqlLibraryOperators.REGEXP_CONTAINS, "regex_match"), FunctionMappings.s(SqlStdOperatorTable.REPLACE, "replace"), FunctionMappings.s(SqlLibraryOperators.REGEXP_REPLACE_3, "regexp_replace"), diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java index 354f797e1beca..d78dfc9eb3cf8 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/JsonFunctionAdapters.java @@ -182,4 +182,38 @@ static class JsonExtendAdapter extends AbstractNameMappingAdapter { super(LOCAL_JSON_EXTEND_OP, List.of(), List.of()); } } + + /** + * {@code JSON_VALID(str)} → boolean; TRUE iff the input parses as JSON, + * FALSE on malformed input, NULL on NULL input. + * + *

Source PPL call uses {@code SqlStdOperatorTable.IS_JSON_VALUE} — a + * {@link org.apache.calcite.sql.SqlPostfixOperator} named {@code "IS JSON VALUE"}, + * which isthmus cannot serialise (no Substrait mapping) and DataFusion does + * not recognise. The adapter rewrites it to a locally-declared {@code json_valid} + * {@link SqlFunction} whose name matches the Rust UDF at {@code rust/src/udf/json_valid.rs}. + * + *

Null-propagating, matching Calcite {@code JsonFunctions.isJsonValue} + * ({@code if (input == null) return null}) and the official PPL doc + * (sql/docs/user/ppl/functions/json.md — "NULL input returns NULL"). This + * is the SQL:2016 scalar-UDF convention and the majority industry contract + * (MySQL, SQL Server, Snowflake, Trino, DuckDB). Return type + * {@link ReturnTypes#BOOLEAN_NULLABLE} matches the postfix operator's declared + * type so {@link AbstractNameMappingAdapter#adapt} preserves it unchanged. + */ + static class JsonValidAdapter extends AbstractNameMappingAdapter { + + static final SqlOperator LOCAL_JSON_VALID_OP = new SqlFunction( + "json_valid", + SqlKind.OTHER_FUNCTION, + ReturnTypes.BOOLEAN_NULLABLE, + null, + OperandTypes.STRING, + SqlFunctionCategory.STRING + ); + + JsonValidAdapter() { + super(LOCAL_JSON_VALID_OP, List.of(), List.of()); + } + } } diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml index fb5b76231421c..d3ab31851ec7e 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml @@ -833,6 +833,12 @@ scalar_functions: variadic: { min: 1 } return: string + - name: "json_valid" + description: "TRUE if the input parses as JSON, FALSE on malformed input, NULL on NULL input (parity with legacy JsonUtils.isValidJson / Jackson readTree — empty/whitespace input is valid)." + impls: + - args: [{ value: string, name: "value" }] + return: boolean + - name: "opensearch_extract" description: >- Pull a MySQL-style calendar component out of a timestamp. Returns BIGINT. diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/JsonFunctionAdaptersTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/JsonFunctionAdaptersTests.java index 7ea4f3550100b..f5c13a12bd9f2 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/JsonFunctionAdaptersTests.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/JsonFunctionAdaptersTests.java @@ -115,4 +115,72 @@ public void testJsonArrayLengthPreservesOriginalReturnType() { adapted.getType() ); } + + // ── JsonValidAdapter ────────────────────────────────────────────────── + + public void testJsonValidRewritesToLocalOp() { + // Synthesize JSON_VALID(value). In production the source operator is + // SqlStdOperatorTable.IS_JSON_VALUE (a SqlPostfixOperator) but the + // adapter's contract is purely shape-based — any single-VARCHAR-operand + // RexCall must rewrite to LOCAL_JSON_VALID_OP. Using a SqlFunction stand-in + // exercises the same code path and keeps the test self-contained. + RelDataType varcharNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.VARCHAR), true); + RelDataType booleanNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BOOLEAN), true); + SqlFunction pplJsonValidOp = new SqlFunction( + "JSON_VALID", + SqlKind.OTHER_FUNCTION, + ReturnTypes.explicit(booleanNullable), + null, + OperandTypes.STRING, + SqlFunctionCategory.STRING + ); + RexNode valueRef = rexBuilder.makeInputRef(varcharNullable, 0); + RexCall original = (RexCall) rexBuilder.makeCall(pplJsonValidOp, List.of(valueRef)); + + RexNode adapted = new JsonFunctionAdapters.JsonValidAdapter().adapt(original, List.of(), cluster); + + assertTrue("adapted node must be a RexCall, got " + adapted.getClass(), adapted instanceof RexCall); + RexCall call = (RexCall) adapted; + assertSame( + "adapted call must target LOCAL_JSON_VALID_OP", + JsonFunctionAdapters.JsonValidAdapter.LOCAL_JSON_VALID_OP, + call.getOperator() + ); + assertEquals("json_valid is unary — no prepend / append", 1, call.getOperands().size()); + assertSame("arg 0 must be the original value operand", valueRef, call.getOperands().get(0)); + } + + /** + * Same regression guard as {@link #testJsonArrayLengthPreservesOriginalReturnType}: + * the adapted call must keep the original call's {@link RelDataType} instance so + * the cached {@code Project.rowType} matches post-adaptation. Production + * {@code IS_JSON_VALUE} returns {@code BOOLEAN_NULLABLE}, same as + * {@code LOCAL_JSON_VALID_OP}, so a naive {@code rexBuilder.makeCall(op, args)} + * would happen to produce the right type — pick a differently-nullable BOOLEAN + * here to make the assertion actually distinguish "preserve" from "infer". + */ + public void testJsonValidPreservesOriginalReturnType() { + RelDataType varcharNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.VARCHAR), true); + RelDataType booleanNotNull = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BOOLEAN), false); + SqlFunction pplJsonValidOp = new SqlFunction( + "JSON_VALID", + SqlKind.OTHER_FUNCTION, + ReturnTypes.explicit(booleanNotNull), + null, + OperandTypes.STRING, + SqlFunctionCategory.STRING + ); + RexNode valueRef = rexBuilder.makeInputRef(varcharNullable, 0); + RexCall original = (RexCall) rexBuilder.makeCall(pplJsonValidOp, List.of(valueRef)); + assertEquals(booleanNotNull, original.getType()); + + RexNode adapted = new JsonFunctionAdapters.JsonValidAdapter().adapt(original, List.of(), cluster); + + assertEquals( + "adapted call's return type must equal the original call's return type, " + + "otherwise the enclosing Project.rowType assertion fails in fragment conversion", + original.getType(), + adapted.getType() + ); + } }