[FLINK-40089][table] Implement JSON_LENGTH function

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()

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
 ----------------------------

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":

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.
+     *
+     * <p>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.
+     *
+     * <p>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.

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
     // --------------------------------------------------------------------------------------------

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java

@@ -84,6 +84,7 @@ Stream<TestSetSpec> getTestSetSpecs() {
         final List<TestSetSpec> 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<TestSetSpec> 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)

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();

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;
+        }
+    }
+}