This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new dc256f305a4 [FLINK-40268][table] Add new JSON_TYPE built-in function
dc256f305a4 is described below
commit dc256f305a444910ee87db648ee4e27207f9bef6
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Fri Aug 14 13:58:44 2026 +0100
[FLINK-40268][table] Add new JSON_TYPE built-in function
This closes #28850.
---
docs/data/sql_functions.yml | 44 ++++++
docs/data/sql_functions_zh.yml | 44 ++++++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 37 +++++
.../flink/table/api/internal/BaseExpressions.java | 48 +++++++
.../functions/BuiltInFunctionDefinitions.java | 20 +++
.../strategies/JsonPathInputTypeStrategy.java | 84 ++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 5 +
.../table/planner/codegen/JsonCodeGenUtils.java | 119 ++++++++++++++++
.../table/planner/codegen/ExprCodeGenerator.scala | 3 +
.../planner/codegen/calls/BuiltInMethods.scala | 11 ++
.../table/planner/codegen/JsonParseReuseTest.java | 49 +++++++
.../planner/functions/JsonFunctionsITCase.java | 150 +++++++++++++++++++++
.../table/runtime/functions/SqlJsonUtils.java | 80 +++++++++++
14 files changed, 695 insertions(+)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 75d85543051..19a50dad2f0 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1267,6 +1267,50 @@ json:
-- JSON_EXISTS separates an absent path from a present one
SELECT JSON_EXISTS(json_doc, '$.items[*]'), JSON_LENGTH(json_doc,
'$.items[*]');
+ - sql: JSON_TYPE(jsonValue[, path])
+ table: jsonType(jsonValue[, path])
+ description: |
+ Returns a string value indicating the type of jsonValue, optionally at
the given JSON
+ `path`. Returns SQL `NULL` if jsonValue is SQL `NULL`, is not valid
JSON, or the path
+ does not resolve to exactly one value.
+
+ The possible results are `object`, `array`, `string`, `number`,
`boolean` and `null` for
+ the JSON null literal.
+
+ Every number is reported as `number`,whatever its magnitude or precision.
+ Anything quoted is a `string`, including dates and date-times.
+
+ ```sql
+ -- object
+ SELECT JSON_TYPE('{"a": true}')
+
+ -- array
+ SELECT JSON_TYPE('[1, 2]')
+
+ -- null
+ SELECT JSON_TYPE('null')
+
+ -- boolean
+ SELECT JSON_TYPE('true')
+
+ -- string
+ SELECT JSON_TYPE('"Hello, World!"')
+
+ -- string
+ SELECT JSON_TYPE('"2015-01-01"')
+
+ -- number
+ SELECT JSON_TYPE('66')
+
+ -- number
+ SELECT JSON_TYPE('11.1')
+
+ -- NULL, since 68s is not valid JSON
+ SELECT JSON_TYPE('68s')
+
+ -- array, read at the given path
+ SELECT JSON_TYPE('{"a": [1, 2]}', '$.a')
+ ```
variant:
- sql: PARSE_JSON(json_string[, allow_duplicate_keys])
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 5b332fca088..613c561d838 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1353,6 +1353,50 @@ json:
-- JSON_EXISTS separates an absent path from a present one
SELECT JSON_EXISTS(json_doc, '$.items[*]'), JSON_LENGTH(json_doc,
'$.items[*]');
+ - sql: JSON_TYPE(jsonValue[, path])
+ table: jsonType(jsonValue[, path])
+ description: |
+ Returns a string value indicating the type of jsonValue, optionally at
the given JSON
+ `path`. Returns SQL `NULL` if jsonValue is SQL `NULL`, is not valid
JSON, or the path
+ does not resolve to exactly one value.
+
+ The possible results are `object`, `array`, `string`, `number`,
`boolean` and `null` for
+ the JSON null literal.
+
+ Every number is reported as `number`,whatever its magnitude or precision.
+ Anything quoted is a `string`, including dates and date-times.
+
+ ```sql
+ -- object
+ SELECT JSON_TYPE('{"a": true}')
+
+ -- array
+ SELECT JSON_TYPE('[1, 2]')
+
+ -- null
+ SELECT JSON_TYPE('null')
+
+ -- boolean
+ SELECT JSON_TYPE('true')
+
+ -- string
+ SELECT JSON_TYPE('"Hello, World!"')
+
+ -- string
+ SELECT JSON_TYPE('"2015-01-01"')
+
+ -- number
+ SELECT JSON_TYPE('66')
+
+ -- number
+ SELECT JSON_TYPE('11.1')
+
+ -- NULL, since 68s is not valid JSON
+ SELECT JSON_TYPE('68s')
+
+ -- array, read at the given path
+ SELECT JSON_TYPE('{"a": [1, 2]}', '$.a')
+ ```
variant:
- sql: PARSE_JSON(json_string[, allow_duplicate_keys])
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst
b/flink-python/docs/reference/pyflink.table/expressions.rst
index 17a475516b9..10614e996b1 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -327,6 +327,7 @@ JSON functions
Expression.json_quote
Expression.json_unquote
Expression.json_length
+ Expression.json_type
value modification functions
----------------------------
diff --git a/flink-python/pyflink/table/expression.py
b/flink-python/pyflink/table/expression.py
index d4a9186882f..33fd03122f8 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -2324,6 +2324,43 @@ class Expression(Generic[T]):
else:
return _binary_op("jsonLength")(self, path)
+ def json_type(self, path=None) -> 'Expression':
+ """
+ Returns a string value indicating the type of the input.
+
+ Potential outputs are as following
+
+ * `object`
+ * `array`
+ * `string`
+ * `number`
+ * `boolean`
+ * `null`, for the JSON null literal
+
+ Every number is reported as `number`, whatever its magnitude or
precision.
+
+ Returns None if the input is None or is not valid JSON.
+
+ If `path` is given, the type is read at that location instead of the
root. A path that
+ does not resolve to exactly one value returns None.
+
+ Examples:
+ ::
+
+ >>> lit('{"a": true}').json_type() # 'object'
+ >>> lit('[1, 2]').json_type() # 'array'
+ >>> lit('null').json_type() # 'null'
+ >>> lit('"Hello, World!"').json_type() # 'string'
+ >>> lit('"2015-01-01"').json_type() # 'string'
+ >>> lit('66').json_type() # 'number'
+ >>> lit('11.1').json_type() # 'number'
+ >>> lit('68s').json_type() # None, not valid JSON
+ >>> lit('{"a": [1, 2]}').json_type('$.a') # 'array'
+ """
+ if path is None:
+ return _unary_op("jsonType")(self)
+ else:
+ return _binary_op("jsonType")(self, path)
# ---------------------------- value modification functions
-----------------------------
def object_update(self, *kv) -> "Expression":
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 53c05ff3c29..1dfc2687cbf 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -143,6 +143,7 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_E
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_LENGTH;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_QUERY;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_QUOTE;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_TYPE;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_UNQUOTE;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_VALUE;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.LAST_VALUE;
@@ -2507,6 +2508,53 @@ public abstract class BaseExpressions<InType, OutType> {
return jsonQuery(path, DataTypes.STRING(), wrappingBehavior, onEmpty,
onError);
}
+ /**
+ * Returns a string value indicating the type of the input.
+ *
+ * <p>Potential outputs are as following
+ *
+ * <ul>
+ * <li>object
+ * <li>array
+ * <li>string
+ * <li>number
+ * <li>boolean
+ * <li>null, for the JSON null literal
+ * </ul>
+ *
+ * <p>Every number is reported as number, whatever its magnitude or
precision.
+ *
+ * <p>Returns SQL NULL if the input is NULL or is not valid JSON.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * lit("{\"a\": true}").jsonType() // "object"
+ * lit("[1, 2]").jsonType() // "array"
+ * lit("null").jsonType() // "null"
+ * lit("\"Hello, World!\"").jsonType() // "string"
+ * lit("\"2015-01-01\"").jsonType() // "string"
+ * lit("66").jsonType() // "number"
+ * lit("11.1").jsonType() // "number"
+ * lit("68s").jsonType() // SQL NULL, not valid JSON
+ * }</pre>
+ */
+ public OutType jsonType() {
+ return toApiSpecificExpression(unresolvedCall(JSON_TYPE, toExpr()));
+ }
+
+ /**
+ * Like {@link #jsonType()}, but reads the type at {@code path} instead of
the root. A path that
+ * does not resolve to exactly one value returns SQL NULL.
+ *
+ * <pre>{@code
+ * lit("{\"a\": [1, 2]}").jsonType("$.a") // "array"
+ * }</pre>
+ */
+ public OutType jsonType(String path) {
+ return toApiSpecificExpression(unresolvedCall(JSON_TYPE, toExpr(),
valueLiteral(path)));
+ }
+
/**
* Extracts JSON values from a JSON string.
*
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 707c6ea5bfe..85edd2ead05 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -117,6 +117,7 @@ import static
org.apache.flink.table.types.inference.strategies.SpecificInputTyp
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.TWO_FULLY_COMPARABLE;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentage;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentageArray;
+import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.plainJsonPath;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.ARRAY_APPEND_PREPEND;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY;
@@ -3102,6 +3103,25 @@ public final class BuiltInFunctionDefinitions {
.runtimeProvided()
.build();
+ public static final BuiltInFunctionDefinition JSON_TYPE =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("JSON_TYPE")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ plainJsonPath(
+ or(
+
sequence(logical(LogicalTypeFamily.CHARACTER_STRING)),
+ sequence(
+
logical(LogicalTypeFamily.CHARACTER_STRING),
+ and(
+ logical(
+
LogicalTypeFamily
+
.CHARACTER_STRING),
+ LITERAL)))))
+ .outputTypeStrategy(explicit(DataTypes.STRING()))
+ .runtimeProvided()
+ .build();
+
//
--------------------------------------------------------------------------------------------
// Variant functions
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/JsonPathInputTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/JsonPathInputTypeStrategy.java
new file mode 100644
index 00000000000..5a7ef1f0a6c
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/JsonPathInputTypeStrategy.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+/**
+ * Strategy for a JSON function whose optional second argument is a path
literal. Rejects the {@code
+ * lax}/{@code strict} path mode prefix at planning time and delegates
everything else to {@code
+ * signatures}.
+ */
+@Internal
+public final class JsonPathInputTypeStrategy implements InputTypeStrategy {
+
+ private static final int ARG_PATH = 1;
+
+ private static final Pattern PATH_MODE_PREFIX =
+ Pattern.compile("\\s*(strict|lax)\\s+\\$.*",
Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
+
+ private final InputTypeStrategy signatures;
+
+ JsonPathInputTypeStrategy(final InputTypeStrategy signatures) {
+ this.signatures = signatures;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return signatures.getArgumentCount();
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean throwOnFailure) {
+ final Optional<List<DataType>> inferredDataTypes =
+ signatures.inferInputTypes(callContext, throwOnFailure);
+ if (inferredDataTypes.isEmpty() ||
callContext.getArgumentDataTypes().size() <= ARG_PATH) {
+ return inferredDataTypes;
+ }
+
+ final Optional<String> path = callContext.getArgumentValue(ARG_PATH,
String.class);
+ if (path.isPresent() &&
PATH_MODE_PREFIX.matcher(path.get()).matches()) {
+ return callContext.fail(
+ throwOnFailure,
+ "%s does not support the 'lax'/'strict' path mode prefix
(got: '%s'). "
+ + "Use a plain path such as '$.a.b'. To check path
existence or handle "
+ + "invalid input, use JSON_EXISTS or IS JSON.",
+ callContext.getName(),
+ path.get());
+ }
+
+ return inferredDataTypes;
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(final FunctionDefinition
definition) {
+ return signatures.getExpectedSignatures(definition);
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index d551ca89a55..9aa75e02463 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -96,6 +96,11 @@ public final class SpecificInputTypeStrategies {
public static final ArgumentTypeStrategy
JSON_QUERY_ON_EMPTY_ERROR_BEHAVIOUR =
new JsonQueryOnErrorEmptyArgumentTypeStrategy();
+ /** See {@link JsonPathInputTypeStrategy}. */
+ public static InputTypeStrategy plainJsonPath(final InputTypeStrategy
signatures) {
+ return new JsonPathInputTypeStrategy(signatures);
+ }
+
/** Argument type derived from the array element type. */
public static final ArgumentTypeStrategy ARRAY_ELEMENT_ARG =
new ArrayElementArgumentTypeStrategy();
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
new file mode 100644
index 00000000000..0b466bded2c
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.codegen;
+
+import org.apache.flink.table.planner.codegen.calls.BuiltInMethods;
+import org.apache.flink.table.runtime.functions.SqlJsonUtils;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import scala.Option;
+import scala.Tuple2;
+import scala.collection.Seq;
+
+/** Utilities for the code generation of JSON functions. */
+public final class JsonCodeGenUtils {
+
+ private JsonCodeGenUtils() {}
+
+ /**
+ * Generates {@code JSON_TYPE(jsonValue)} or {@code JSON_TYPE(jsonValue,
path)}.
+ *
+ * <p>The parsed context is shared with the other JSON functions over the
same input, so the
+ * input is parsed only once per record.
+ */
+ public static GeneratedExpression generateJsonType(
+ CodeGeneratorContext ctx, LogicalType returnType,
Seq<GeneratedExpression> operands) {
+ boolean hasPath = operands.length() == 2;
+ return GenerateUtils.generateCallWithStmtIfArgsNotNull(
+ ctx,
+ returnType,
+ operands,
+ true,
+ false,
+ argTerms -> {
+ ParsedJson parsed = getOrCreateParsedJson(ctx,
argTerms.head() + ".toString()");
+ String call;
+ if (hasPath) {
+ String pathSpec =
operands.apply(1).literalValue().get().toString();
+ boolean definite =
SqlJsonUtils.isPathDefinite(pathSpec);
+ call =
+
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_TYPE_PATH())
+ + "("
+ + parsed.varName
+ + ", "
+ + argTerms.apply(1)
+ + ".toString(), "
+ + definite
+ + ")";
+ } else {
+ call =
+
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_TYPE())
+ + "("
+ + parsed.varName
+ + ")";
+ }
+ String resultExpr = CodeGenUtils.BINARY_STRING() +
".fromString(" + call + ")";
+ return new Tuple2<>(parsed.parseCode, resultExpr);
+ });
+ }
+
+ /**
+ * Emits code that parses the given JSON {@code inputTerm} into a reusable
{@link
+ * SqlJsonUtils.JsonValueContext} member variable.
+ *
+ * @return the parsed-context variable name and the parse statement, empty
if the same input was
+ * already parsed
+ */
+ private static ParsedJson getOrCreateParsedJson(CodeGeneratorContext ctx,
String inputTerm) {
+ Option<GeneratedExpression> existing =
+ ctx.getReusableInputUnboxingExprs(inputTerm,
Integer.MIN_VALUE);
+ if (existing.isDefined()) {
+ return new ParsedJson(existing.get().resultTerm(), "");
+ }
+
+ String varName = CodeGenUtils.newName(ctx, "jsonParsed");
+ String typeName = SqlJsonUtils.JsonValueContext.class.getName();
+ ctx.addReusableMember(typeName + " " + varName + ";");
+
+ ctx.addReusableInputUnboxingExprs(
+ inputTerm,
+ Integer.MIN_VALUE,
+ new GeneratedExpression(varName, "false", "", null,
Option.empty()));
+
+ String parseCode =
+ varName
+ + " = "
+ +
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_PARSE())
+ + "("
+ + inputTerm
+ + ");";
+ return new ParsedJson(varName, parseCode);
+ }
+
+ /** Holds the outcome of {@link #getOrCreateParsedJson}. */
+ private static final class ParsedJson {
+ private final String varName;
+ private final String parseCode;
+
+ ParsedJson(String varName, String parseCode) {
+ this.varName = varName;
+ this.parseCode = parseCode;
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
index 024229f2563..7e5779ea8c6 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
@@ -939,6 +939,9 @@ class ExprCodeGenerator(
case BuiltInFunctionDefinitions.JSON_LENGTH =>
new JsonLengthCallGen().generate(ctx, operands, resultType)
+ case BuiltInFunctionDefinitions.JSON_TYPE =>
+ JsonCodeGenUtils.generateJsonType(ctx, resultType, operands)
+
case BuiltInFunctionDefinitions.INTERNAL_HASHCODE =>
new HashCodeCallGen().generate(ctx, operands, resultType)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
index 6d0d695b39a..99ea6eb483f 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -513,6 +513,17 @@ object BuiltInMethods {
val JSON_PARSE =
Types.lookupMethod(classOf[SqlJsonUtils], "jsonParse", classOf[String])
+ val JSON_TYPE =
+ Types.lookupMethod(classOf[SqlJsonUtils], "jsonType",
classOf[SqlJsonUtils.JsonValueContext])
+
+ val JSON_TYPE_PATH = Types.lookupMethod(
+ classOf[SqlJsonUtils],
+ "jsonType",
+ classOf[SqlJsonUtils.JsonValueContext],
+ classOf[String],
+ classOf[Boolean]
+ )
+
val JSON_QUERY_PARSED = Types.lookupMethod(
classOf[SqlJsonUtils],
"jsonQueryParsed",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
index 9d646ade3aa..abceea8e83c 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
@@ -271,6 +271,55 @@ class JsonParseReuseTest {
.isOne();
}
+ @Test
+ void testSingleJsonTypeCall() {
+ final String sql = "SELECT JSON_TYPE(json_data) FROM json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).containsExactlyInAnyOrder(Row.of("object"),
Row.of("object"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("A single JSON_TYPE call should parse once")
+ .isOne();
+ }
+
+ @Test
+ void testTwoJsonTypeCalls() {
+ final String sql = "SELECT JSON_TYPE(json_data), JSON_TYPE(json_data)
FROM json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows)
+ .containsExactlyInAnyOrder(Row.of("object", "object"),
Row.of("object", "object"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("Identical JSON_TYPE calls are one expression, so they
parse once")
+ .isOne();
+ }
+
+ @Test
+ void testJsonTypeAndJsonValueMixed() {
+ final String sql =
+ "SELECT JSON_VALUE(json_data, '$.type'), JSON_TYPE(json_data)
FROM json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows)
+ .containsExactlyInAnyOrder(Row.of("account", "object"),
Row.of("admin", "object"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("JSON_VALUE + JSON_TYPE on the same input should parse
once")
+ .isOne();
+ }
+
+ @Test
+ void testJsonTypeWithJsonValueAndJsonQuery() {
+ final String sql =
+ "SELECT JSON_VALUE(json_data, '$.type'), "
+ + "JSON_QUERY(json_data, '$.address'), "
+ + "JSON_TYPE(json_data) FROM json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows)
+ .containsExactlyInAnyOrder(
+ Row.of("account", "{\"city\":\"Munich\"}", "object"),
+ Row.of("admin", "{\"city\":\"Berlin\"}", "object"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("JSON_VALUE + JSON_QUERY + JSON_TYPE on the same input
should parse once")
+ .isOne();
+ }
+
@Test
void testDifferentJsonInputs() {
final String sql =
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
index 9f3b1e377c3..65ffef63b2d 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
@@ -94,6 +94,7 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
testCases.addAll(jsonObjectSpec());
testCases.addAll(jsonSpec());
testCases.addAll(jsonArraySpec());
+ testCases.addAll(jsonTypeSpec());
testCases.addAll(jsonQuoteSpec());
testCases.addAll(jsonUnquoteSpecWithValidInput());
testCases.addAll(jsonUnquoteSpecWithInvalidInput());
@@ -1423,6 +1424,155 @@ class JsonFunctionsITCase extends
BuiltInFunctionTestBase {
STRING().notNull()));
}
+ private static List<TestSetSpec> jsonTypeSpec() {
+ return List.of(
+ // One flag per JSON type.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData(
+ "{\"a\": true}", "[1, 2]", "true", "\"Hello,
World!\"", "66")
+ .andDataTypes(STRING(), STRING(), STRING(), STRING(),
STRING())
+ .testResult(
+ $("f0").jsonType(), "JSON_TYPE(f0)", "object",
STRING().nullable())
+ .testResult(
+ $("f1").jsonType(), "JSON_TYPE(f1)", "array",
STRING().nullable())
+ .testResult(
+ $("f2").jsonType(), "JSON_TYPE(f2)",
"boolean", STRING().nullable())
+ .testResult(
+ $("f3").jsonType(), "JSON_TYPE(f3)", "string",
STRING().nullable())
+ .testResult(
+ $("f4").jsonType(), "JSON_TYPE(f4)", "number",
STRING().nullable()),
+
+ // The flag follows the JSON grammar alone: a number has no
width, and a quoted
+ // value is a string whatever it spells.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData(
+ "11.1", "99999999999999999999",
"\"2015-01-01\"", "\"66\"")
+ .andDataTypes(STRING(), STRING(), STRING(), STRING())
+ .testResult(
+ $("f0").jsonType(), "JSON_TYPE(f0)", "number",
STRING().nullable())
+ .testResult(
+ $("f1").jsonType(), "JSON_TYPE(f1)", "number",
STRING().nullable())
+ .testResult(
+ $("f2").jsonType(), "JSON_TYPE(f2)", "string",
STRING().nullable())
+ .testResult(
+ $("f3").jsonType(), "JSON_TYPE(f3)", "string",
STRING().nullable()),
+
+ // A SQL NULL input and invalid JSON both yield SQL NULL; the
JSON null literal
+ // returns the string 'null'.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData("68s", "null")
+ .andDataTypes(STRING(), STRING())
+ .testResult(
+ nullOf(STRING()).jsonType(),
+ "JSON_TYPE(CAST(NULL AS STRING))",
+ null,
+ STRING().nullable())
+ .testResult($("f0").jsonType(), "JSON_TYPE(f0)", null,
STRING().nullable())
+ .testResult(
+ $("f1").jsonType(), "JSON_TYPE(f1)", "null",
STRING().nullable())
+ .testResult(
+ $("f1").jsonType("$"),
+ "JSON_TYPE(f1, '$')",
+ "null",
+ STRING().nullable())
+ .testResult(
+ $("f1").jsonType("$.a"),
+ "JSON_TYPE(f1, '$.a')",
+ null,
+ STRING().nullable()),
+
+ // A path reads the type at that location instead of the root,
and yields NULL
+ // unless it resolves to exactly one value. A wildcard path is
indefinite: it reads
+ // back as a list, so it has a type only for a single match.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData("{\"a\": [1, 2]}", "{\"a\": [1]}")
+ .andDataTypes(STRING(), STRING())
+ .testResult(
+ $("f0").jsonType("$.a"),
+ "JSON_TYPE(f0, '$.a')",
+ "array",
+ STRING().nullable())
+ .testResult(
+ $("f0").jsonType("$.a[0]"),
+ "JSON_TYPE(f0, '$.a[0]')",
+ "number",
+ STRING().nullable())
+ .testResult(
+ $("f0").jsonType("$.b"),
+ "JSON_TYPE(f0, '$.b')",
+ null,
+ STRING().nullable())
+ .testResult(
+ $("f0").jsonType("$.["),
+ "JSON_TYPE(f0, '$.[')",
+ null,
+ STRING().nullable())
+ .testResult(
+ $("f0").jsonType(""),
+ "JSON_TYPE(f0, '')",
+ null,
+ STRING().nullable())
+ .testResult(
+ $("f0").jsonType("$.a[*]"),
+ "JSON_TYPE(f0, '$.a[*]')",
+ null,
+ STRING().nullable())
+ .testResult(
+ $("f1").jsonType("$.a[*]"),
+ "JSON_TYPE(f1, '$.a[*]')",
+ "number",
+ STRING().nullable()),
+
+ // The 'lax'/'strict' path mode prefix is rejected at planning
time, but a field of
+ // that name is addressed like any other.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData(
+ "{\"a\": 1}", "{\"lax\": {\"strict\": 2},
\"strict value\": 1}")
+ .andDataTypes(STRING(), STRING())
+ .testSqlValidationError(
+ "JSON_TYPE(f0, 'lax $.a')",
+ "JSON_TYPE does not support the 'lax'/'strict'
path mode prefix "
+ + "(got: 'lax $.a'). Use a plain path
such as '$.a.b'. To "
+ + "check path existence or handle
invalid input, use "
+ + "JSON_EXISTS or IS JSON.")
+ .testTableApiValidationError(
+ $("f0").jsonType("strict $.a"),
+ "JSON_TYPE does not support the 'lax'/'strict'
path mode prefix "
+ + "(got: 'strict $.a'). Use a plain
path such as '$.a.b'. "
+ + "To check path existence or handle
invalid input, use "
+ + "JSON_EXISTS or IS JSON.")
+ .testResult(
+ $("f1").jsonType("lax"),
+ "JSON_TYPE(f1, 'lax')",
+ "object",
+ STRING().nullable())
+ .testResult(
+ $("f1").jsonType("$[\"strict value\"]"),
+ "JSON_TYPE(f1, '$[\"strict value\"]')",
+ "number",
+ STRING().nullable()),
+
+ // Only CHARACTER_STRING casts implicitly to VARCHAR, so a
non-string is rejected
+ // rather than coerced. A path argument must be a literal.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_TYPE)
+ .onFieldsWithData(1, "{}")
+ .andDataTypes(INT(), STRING())
+ .testTableApiValidationError(
+ $("f0").jsonType(),
+ "Invalid input arguments. Expected signatures
are:\n"
+ + "JSON_TYPE(<CHARACTER_STRING>)")
+ .testSqlValidationError(
+ "JSON_TYPE(f0)",
+ "Invalid input arguments. Expected signatures
are:\n"
+ + "JSON_TYPE(<CHARACTER_STRING>)")
+ .testSqlValidationError("JSON_TYPE(f1, f1)", "Invalid
input arguments.")
+ .testSqlValidationError(
+ "JSON_TYPE()",
+ "No match found for function signature
JSON_TYPE().\n"
+ + "Supported signatures are:\n"
+ + "JSON_TYPE(<CHARACTER_STRING>)"));
+ }
+
private static List<TestSetSpec> jsonQuoteSpec() {
return List.of(
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
index 6b77cc4fa22..e921cca0324 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
@@ -97,6 +97,12 @@ public class SqlJsonUtils {
private static final String JSON_VALUE_FUNCTION_NAME = "JSON_VALUE";
private static final String JSON_EXISTS_FUNCTION_NAME = "JSON_EXISTS";
+ private static final Configuration JSON_PATH_TYPE_CONFIG =
+ Configuration.builder()
+ .jsonProvider(JSON_PATH_JSON_PROVIDER)
+ .mappingProvider(JSON_PATH_MAPPING_PROVIDER)
+ .build();
+
private SqlJsonUtils() {}
static {
@@ -526,6 +532,80 @@ public class SqlJsonUtils {
return JSON_PATH_JSON_PROVIDER.parse(input);
}
+ /**
+ * Returns the JSON type flag for the parsed value: {@code object}, {@code
array}, {@code
+ * string}, {@code number}, {@code boolean}, or {@code null} for the JSON
null literal. Returns
+ * SQL {@code NULL} for invalid JSON.
+ */
+ public static String jsonType(final JsonValueContext parsedInput) {
+ // Unparsed, or shared with a call that never assigned it: report NULL
either way.
+ if (parsedInput == null || parsedInput.hasException()) {
+ return null;
+ }
+ return getJsonType(parsedInput.obj);
+ }
+
+ private static String getJsonType(final Object val) {
+ if (val instanceof Number) {
+ return "number";
+ } else if (val instanceof String) {
+ return "string";
+ } else if (val instanceof Boolean) {
+ return "boolean";
+ } else if (val instanceof Map) {
+ return "object";
+ } else if (val instanceof Collection) {
+ return "array";
+ } else if (val == null) {
+ return "null";
+ }
+ return null;
+ }
+
+ /**
+ * Returns the JSON type flag at {@code path}, or {@code null} if the path
doesn't resolve to
+ * exactly one value. {@code definite} is computed at plan time by {@link
#isPathDefinite}.
+ */
+ public static String jsonType(
+ final JsonValueContext parsedInput, final String path, final
boolean definite) {
+ if (parsedInput == null || parsedInput.hasException() ||
path.isEmpty()) {
+ return null;
+ }
+
+ if (parsedInput.obj == null) {
+ return "$".equals(path) ? "null" : null;
+ }
+
+ final Object value;
+ try {
+ // PathNotFoundException extends InvalidPathException (covers both
exceptions)
+ value = JsonPath.parse(parsedInput.obj,
JSON_PATH_TYPE_CONFIG).read(path);
+ } catch (InvalidPathException e) {
+ return null;
+ }
+
+ if (!definite) {
+ // Indefinite paths (e.g. wildcards) read back as a list; only one
match has one type.
+ final List<?> matched = (List<?>) value;
+ return matched.size() == 1 ? getJsonType(matched.get(0)) : null;
+ }
+ return getJsonType(value);
+ }
+
+ /** Returns whether {@code pathSpec} is a definite JSON path. */
+ public static boolean isPathDefinite(final String pathSpec) {
+ // JsonPath.compile() rejects an empty path with an
IllegalArgumentException rather than
+ // the InvalidPathException caught below.
+ if (pathSpec.isEmpty()) {
+ return false;
+ }
+ try {
+ return JsonPath.isPathDefinite(pathSpec);
+ } catch (InvalidPathException e) {
+ return false;
+ }
+ }
+
/**
* Parses a JSON string into a reusable context object. The result can be
passed to {@link
* #jsonValue} or {@link #jsonQueryParsed} to avoid re-parsing the same
JSON string multiple
