diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml index 1535c4d375f0e..8d496e6af9794 100644 --- a/docs/data/sql_functions.yml +++ b/docs/data/sql_functions.yml @@ -1685,6 +1685,17 @@ bitmapagg: `bitmap BITMAP` Returns a `BIGINT`. + - sql: JSON_LENGTH(json_doc[, path]) + table: jsonLength(jsonObject[, path]) + description: | + Returns the number of elements in a JSON document, or the length of the value at the specified path if one is provided. + Returns NULL if the argument is NULL or the path does not locate a value. + + The length is determined as follows: + a scalar value (number, string, boolean, or null) has length 1; + an array has a length equal to the number of its elements; + and an object has a length equal to the number of its key-value pairs. + Nested arrays and objects each count as a single element and their contents are not included in the count. catalog: - sql: CURRENT_DATABASE() diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst index e2aeb34f41107..afa67f2a8a48e 100644 --- a/flink-python/docs/reference/pyflink.table/expressions.rst +++ b/flink-python/docs/reference/pyflink.table/expressions.rst @@ -319,13 +319,13 @@ JSON functions .. autosummary:: :toctree: api/ - Expression.is_json Expression.json_exists Expression.json_value Expression.json_query Expression.json_quote Expression.json_unquote + Expression.json_length value modification functions ---------------------------- diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py index 82c49f2edc63d..f764739fdcc8d 100644 --- a/flink-python/pyflink/table/expression.py +++ b/flink-python/pyflink/table/expression.py @@ -2257,6 +2257,23 @@ def json_unquote(self) -> 'Expression': """ return _unary_op("jsonUnquote")(self) + def json_length(self, path = None) -> 'Expression': + """ + Return the length of a JSON value, or of the value at `path` if given. + + - Scalars have length 1. + - Arrays have length equal to their number of elements. + - Objects have length equal to their number of keys. + - Nested arrays/objects are not counted recursively. + - Returns None if the value is None. + """ + + + return _unary_op("jsonLength")(self) if path is None else _binary_op("jsonLength")(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 074224ae696fd..81e84bab81579 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 @@ -140,6 +140,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_TRUE; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_VALID_UTF8; import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_EXISTS; +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_UNQUOTE; @@ -2523,6 +2524,27 @@ public OutType jsonQuery(String path, JsonQueryWrapper wrappingBehavior) { return jsonQuery( path, wrappingBehavior, JsonQueryOnEmptyOrError.NULL, JsonQueryOnEmptyOrError.NULL); } + /** + * Returns the number of elements contained in a JSON value, optionally at a given path. + * + *

Counting works differently depending on the JSON type: objects report how many + * key-value pairs they contain, arrays report how many entries they hold, and any scalar + * value (such as a number, string, or boolean) is treated as a single element and reports 1. + * Only the top level is measured — elements that are themselves arrays or objects contribute + * 1 to the count regardless of what they contain. + * + *

When a path is provided, the count applies to the value found at that path rather than + * the document as a whole. Returns NULL if any argument is NULL{@code} or the path does not + * match a value. + */ + + public OutType jsonLength() { + return toApiSpecificExpression(unresolvedCall(JSON_LENGTH, toExpr())); + } + + public OutType jsonLength(String path) { + return toApiSpecificExpression(unresolvedCall(JSON_LENGTH, 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 0828f2bb8b4e9..d7511fe36e35a 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 @@ -3081,6 +3081,23 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL) .runtimeProvided() .build(); + public static final BuiltInFunctionDefinition JSON_LENGTH = + BuiltInFunctionDefinition.newBuilder() + .name("JSON_LENGTH") + .kind(SCALAR) + .inputTypeStrategy( + or( + sequence(logical(LogicalTypeFamily.CHARACTER_STRING)), + sequence( + logical(LogicalTypeFamily.CHARACTER_STRING), + and( + logical(LogicalTypeFamily.CHARACTER_STRING), + LITERAL)))) + .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.INT()))) + .runtimeClass( + "org.apache.flink.table.runtime.functions.scalar.JsonLengthFunction") + .build(); + // I need to complete this // -------------------------------------------------------------------------------------------- // Variant functions // -------------------------------------------------------------------------------------------- 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 89a0448e95eef..47ac3f928be68 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 @@ -84,6 +84,7 @@ Stream getTestSetSpecs() { final List testCases = new ArrayList<>(); testCases.add(jsonExistsSpec()); testCases.add(jsonValueSpec()); + testCases.add(jsonLengthSpec()); testCases.addAll(isJsonSpec()); testCases.addAll(jsonQuerySpec()); testCases.addAll(jsonStringSpec()); @@ -97,6 +98,58 @@ Stream getTestSetSpecs() { return testCases.stream(); } + private static TestSetSpec jsonLengthSpec() { + final String jsonValue = getJsonFromResource("/json/json-exists.json"); + + return TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_LENGTH) + .onFieldsWithData( + jsonValue, // f0: existing resource JSON + "{\"a\":1,\"b\":2}", // f1: object + "[1,2,3]", // f2: array + "\"abc\"", // f3: scalar string + "null", // f4: JSON null + "{", // f5: invalid JSON + (String) null) // f6: SQL NULL + .andDataTypes( + STRING(), + STRING(), + STRING(), + STRING(), + STRING(), + STRING(), + STRING()) + + // SQL NULL input + .testSqlResult("JSON_LENGTH(f6)", null, INT().nullable()) + + // whole-document length from the existing resource: + // top-level keys are type, author, metadata + .testSqlResult("JSON_LENGTH(f0)", 3, INT().nullable()) + + // basic shapes + .testSqlResult("JSON_LENGTH(f1)", 2, INT().nullable()) + .testSqlResult("JSON_LENGTH(f2)", 3, INT().nullable()) + .testSqlResult("JSON_LENGTH(f3)", 1, INT().nullable()) + + // keep this line aligned with your intended semantics + // current implementation returns NULL for JSON literal null + .testSqlResult("JSON_LENGTH(f4)", null, INT().nullable()) + // if you later treat JSON null as a scalar, change expected result to 1 + + // invalid JSON -> NULL + .testSqlResult("JSON_LENGTH(f5)", null, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$')", 3, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.type')", 1, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.author')", 2, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.author.address')", 2, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.metadata.tags')", 3, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.metadata.references')", 1, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.metadata.references[0]')", 2, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.metadata.references[0].url')", 1, INT().nullable()) + .testSqlResult("JSON_LENGTH(f0, '$.missing')", null, INT().nullable()); + } + + private static TestSetSpec jsonExistsSpec() { final String jsonValue = getJsonFromResource("/json/json-exists.json"); return TestSetSpec.forFunction(BuiltInFunctionDefinitions.JSON_EXISTS) 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 c73e998133996..0bfc149885e91 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 @@ -374,6 +374,27 @@ private static Object errorResultForJsonQuery( } } + public static Integer jsonLength(String input, String pathSpec) { + return jsonLength(jsonApiCommonSyntax(input, pathSpec)); + } + + private static Integer jsonLength(JsonPathContext context) { + if (context.hasException()) { + return null; + } + final Object value = context.obj; + if (value == null) { + return null; + } + if (value instanceof Map) { + return ((Map) value).size(); + } + if (value instanceof Collection) { + return ((Collection) value).size(); + } + return 1; + } + public static Object json(String input) { try { String trimmed = input.trim(); diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/JsonLengthFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/JsonLengthFunction.java new file mode 100644 index 0000000000000..365d848e2ad26 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/JsonLengthFunction.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.runtime.functions.scalar; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext; +import org.apache.flink.table.runtime.functions.SqlJsonUtils; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import javax.annotation.Nullable; + +/** Implementation of {@link BuiltInFunctionDefinitions#JSON_LENGTH}. */ +@Internal +public class JsonLengthFunction extends BuiltInScalarFunction { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + public JsonLengthFunction(SpecializedContext context) { + super(BuiltInFunctionDefinitions.JSON_LENGTH, context); + } + + public @Nullable Integer eval(@Nullable StringData jsonInput) { + if (jsonInput == null) { + return null; + } + final JsonNode jsonNode = parse(jsonInput.toString()); + if (jsonNode == null) { + return null; + } + final int length = computeLength(jsonNode); + return length == -1 ? null : length; + } + + public @Nullable Integer eval(@Nullable StringData jsonInput, @Nullable StringData path) { + if (jsonInput == null || path == null) { + return null; + } + try { + return SqlJsonUtils.jsonLength(jsonInput.toString(), path.toString()); + } catch (Exception e) { + return null; + } + } + + private static int computeLength(JsonNode jsonNode) { + if (jsonNode.isArray() || jsonNode.isObject()) { + return jsonNode.size(); + } else if (jsonNode.isTextual() + || jsonNode.isNumber() + || jsonNode.isBoolean() + || jsonNode.isBinary()) { + return 1; + } + return -1; + } + + private static @Nullable JsonNode parse(String jsonStr) { + try { + return MAPPER.readTree(jsonStr); + } catch (Exception e) { + return null; + } + } +}