[FLINK-40089][table] Implement JSON_LENGTH function by VasShabu · Pull Request #28688 · apache/flink

VasShabu · 2026-07-08T12:33:59Z

What is the purpose of the change

This pull request adds support for the built-in SQL function JSON_LENGTH, which returns the number of top-level elements in a JSON array or object. Scalars have a length of 1, and NULL input returns NULL. Nested arrays or objects are not counted recursively.

Brief change log

Added JsonLengthFunction implementing the JSON_LENGTH built-in function
Registered JSON_LENGTH as a built-in function in the function catalog
Added test cases for JSON_LENGTH in JsonFunctionsITCase

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

This change added tests and can be verified as follows:

mvn test -pl flink-table/flink-table-runtime -Dtest=JsonFunctionsITCase
Added test cases for JSON_LENGTH covering scalars, arrays, objects, and NULL input in JsonFunctionsITCase

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs (sql_functions.yml)

Was generative AI tooling used to co-author this PR?

Yes (Claude)

flinkbot · 2026-07-08T12:43:54Z

CI report:

7e21cc3 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

raminqaf

Thanks for the contribution @VasShabu! Left some comments!

raminqaf · 2026-07-08T12:47:39Z

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


Why to we need this? Don't we have already the runtime class JsonLengthFunction?

raminqaf · 2026-07-08T12:48:56Z

+        super(BuiltInFunctionDefinitions.JSON_LENGTH, context);
+    }
+
+    public @Nullable Integer eval(@Nullable StringData jsonInput) {


nit: here and everywhere else make parameters/variables that are immutable final (not enforced on Flink but nicer to read).

Suggested change

public @Nullable Integer eval(@Nullable StringData jsonInput) {

public @Nullable Integer eval(final @Nullable StringData jsonInput) {

gustavodemorais

Thanks for the contribution, @VasShabu! Good job on the first version of the PR. We'll go through some iterations to make sure things look good

Apart from what we already flagged above: we have two implementations of the same function. eval(json) uses a new Jackson readTree, eval(json, path) uses the existing Calcite path in SqlJsonUtils. Different parsers + different data models = they can disagree on the same input. We can make the design more consistent by using only one path. Duplicate ObjectMapper - reuse the configured one in SqlJsonUtils instead of a second bare instance. This is important because built-in functions are on the hot path. The logic might be invoked millions of times in a couple of seconds and we have to make sure they allocate as little as possible and their performance are optimized or else the engine will be wasting resources.

That said, In this case, we could go two paths

Best performance: We build a custom parser to optimize it for length checking by skipping the children. MAPPER.getFactory().createParser(str), read first token; if START_ARRAY/START_OBJECT walk top level counting, skipping children via parser.skipChildren(). No tree alloc. Best for length-only.
Simplest reuse the shared SqlJsonUtils path so both overloads share one parser.

Take a look at both paths and let me know what you think

gustavodemorais · 2026-07-08T14:28:37Z

+        } else if (jsonNode.isTextual()
+                || jsonNode.isNumber()
+                || jsonNode.isBoolean()
+                || jsonNode.isBinary()) {


Isn't this check dead code? Can a jsonNode be binary? You can try to write a test to check if it's dead code or it's possible

snuyanzin

Are we sure the selected approach is the right one?
With this approach the same JSON will be parsed several times for queries like

SELECT JSON_LENGTH(my_json), JSON_LENGTH(my_json), JSON_LENGTH(my_json) FROM...

or

SELECT JSON_LENGTH(my_json), JSON_VALUE(my_json, '$.a') FROM...

snuyanzin · 2026-07-08T18:21:32Z

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


this does not look true
simple tests show different behavior

SELECT json_length('$'); SELECT json_length('{');

each returns -1 and the doc says nothing about this or did I miss anything?

snuyanzin

More findings:

Doc doesn't reflect the real behavior: it returns -1 in case of invalid json, while docs says about NULL, e.g. query SELECT JSON_LENGTH('null');
Strange handling of null values
like a query
```
SELECT json_length('{"a":[true, false, null]}', '$.a[1]')
```
both in MySQL and in Flink it returns 1
now
```
SELECT json_length('{"a":[true, false, null]}', '$.a[2]')
```
in MySQL it returns 1, in Flink it returns -1 why?

Seems need to check what the behavior should be in case of lax/strict since right now it is weird

SELECT json_length('{"x": {"a":1}, "y": {"a":[2, 3]}}', 'lax $.*.a');  -- returns 2

SELECT json_length('{"x": {"a":[1, 2, 3, 4]}, "y": {"a":[2, 3]}}', 'lax $.*.a');  -- returns 2

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}, "z":{"a":1}}', 'lax *.a'); -- returns 3

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}, "z":{"a":1}}', 'lax *.a[0]'); --returns 1

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}, "z":{"a":1}}', 'lax *.a[1]'); --returns 1

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}, "z":{"a":1}}', 'lax *.a[2]'); --returns 1

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}, "z":{"a":1}}', 'lax *.a[3]');  --returns 0

davidradl · 2026-07-13T13:57:17Z

        """
        return _unary_op("jsonUnquote")(self)

+    def json_length(self, path = None) -> 'Expression':


It seems that many databases throw an error, I am curious why we are not throwing an error. Also I see that there are best practises that we should document like using IF(string_column IS JSON, to protect against non json and avoid potential errors.

snuyanzin · 2026-07-14T09:26:45Z

+        final JsonPathContext context = jsonApiCommonSyntax(parsedInput, pathSpec);
+        final Object value = context.hasException() ? null : context.obj;
+
+        if (value instanceof LinkedList) {


is there a reason we expect LinkedList instead of just List?

the comment marked as resolved however no answer yet
can you elaborate here?

snuyanzin · 2026-07-14T09:28:13Z

+logger.codelog.name = org.apache.flink.table.runtime.generated.CompileUtils
+logger.codelog.level = DEBUG
+logger.codelog.appenderRef.test.ref = TestLogger


it might be helpful for local debug, however why do we need in repo and making it executable for every ci run?

raminqaf

Another pass

raminqaf · 2026-07-14T10:08:01Z

+            if (matched.size() != 1) {
+                if (context.mode == PathMode.STRICT || matched.isEmpty()) {
+                    return null;
+                }
+                return matched.size();
+            }


Could you please elaborate on this? In which conditions we get a null and in which conditions we get the size of the matched?

looks like not addressed yet

snuyanzin · 2026-07-14T16:21:49Z

+#logger.codelog.name = org.apache.flink.table.runtime.generated.CompileUtils
+#logger.codelog.level = DEBUG
+#logger.codelog.appenderRef.test.ref = TestLogger
+


can we just drop it?

keeping log setting for every helper here might be overkill

snuyanzin · 2026-07-14T16:24:17Z

+            generateCallWithStmtIfArgsNotNull(ctx, resultType, operands, resultNullable = true) {
+              argTerms =>
+                val inputTerm = s"${argTerms.head}.toString()"
+                val (varName, parseCode) =
+                  ctx.getReusableInputUnboxingExprs(inputTerm, Int.MinValue) match {
+                    case Some(expr) => (expr.resultTerm, "")
+                    case None =>
+                      val v = CodeGenUtils.newName(ctx, "jsonParsed")
+                      ctx.addReusableMember(
+                        s"${classOf[SqlJsonUtils.JsonValueContext].getName} $v;")
+                      ctx.addReusableInputUnboxingExprs(
+                        inputTerm,
+                        Int.MinValue,
+                        GeneratedExpression(v, "false", "", null))
+                      (v, s"$v = ${qualifyMethod(BuiltInMethods.JSON_PARSE)}($inputTerm);")
+                  }
+                val (method, terms) =
+                  if (operands.length > 1)
+                    (BuiltInMethods.JSON_LENGTH_PATH, Seq(varName, s"${argTerms(1)}.toString()"))
+                  else
+                    (BuiltInMethods.JSON_LENGTH, Seq(varName))
+                (parseCode, s"${qualifyMethod(method)}(${terms.mkString(", ")})")


is it possible to extract it ideally to java helper?

snuyanzin

it seems the lax problem is still not solved

to highlight the issue there are 2 queries:
1.

SELECT json_length('{"x": {"a":[3, 2 , 3]}}', 'lax *.a');

which returns 3, I guess the size of x.a.
2.

SELECT json_length('{"x": {"a":[3, 2 , 3]}, "y": {"a":null}}', 'lax *.a');

now returns 2 which is amount of x.a, not the amount of elements.

There is nothing about this in doc, and it is very non obvious that depending on query the same syntax will mean completely different things

and there is no info about this in docs

VasShabu · 2026-07-15T16:59:47Z

I have decided to go with the approach of not allowing the use of wildcards, just like how MariaDB chose to implement this because when you have 2 potential paths, the output of the function is ambiguous. you will not know as a user if you have found a path with 5 element or if you pattern matches 5 different paths.

gustavodemorais

I've done a new pass. Thanks for addressing the feedback, Vas. There are still some open topics, make sure to go through all the reviews so we try to converge and close the open points from all the reviewers.

gustavodemorais · 2026-07-17T09:02:50Z

+
+      Nested arrays and objects each count as a single element and their contents are not included in the count.
+
+      When provided with a path that uses a wildcard and resolves in 2 or more paths, JSON_LENGTH will resolve as NULL. 


Make sure to sync the chinese version

gustavodemorais · 2026-07-17T09:04:55Z

        return _unary_op("jsonUnquote")(self)

+    def json_length(self, path = None) -> 'Expression':
+        """


Use the same javadoc and the one you added to the java interface in BaseExpressions.java. We use the Python API as a mirror of the java one and we don't want to have two versions

gustavodemorais · 2026-07-17T09:09:04Z

+     *   <li>Returns Null if the json input is Null or if the given path does not resolve to
+     *       anything.
+     * </ul>
+     *


Examples are always very useful so add an example here as well

gustavodemorais · 2026-07-17T09:14:39Z

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


Suggested change

Returns NULL if the argument is NULL or the path does not locate a value.

Returns NULL if the argument is NULL, the json is invalid, or the path does not locate a value.

gustavodemorais · 2026-07-17T09:16:56Z

+        return toApiSpecificExpression(unresolvedCall(JSON_LENGTH, toExpr()));
+    }
+
+    public OutType jsonLength(String path) {


Add documentation and examples here

gustavodemorais · 2026-07-17T09:27:32Z

+                .testSqlResult("JSON_LENGTH('{\"a\":[1,2,3]}', '$.b')", null, INT().nullable())
+                .testSqlResult("JSON_LENGTH('{\"a\":[1,2,3]}', '$.a')", 3, INT().nullable())
+
+                // lax vs strict


Since we've said both behaviors will be the same, I guess "// lax vs strict" tests are not bringing value and we can delete them?

We should reject lax/strict mode with a nice error stating that we don't support and and point users to use the helper functions "IS JSON/JSON_EXISTS"

gustavodemorais · 2026-07-17T09:39:45Z

+        JSON_LENGTH('hello')
+        -- 1
+
+        JSON_LENGTH('{1: "hello", 2: "bye bye"}', '$.2')


Let's add an example using it with IS_JSON for the case where the user pass on param and JSON_EXISTS for the case with a path. We always want to point users to good patterns, which in this case is using helper functions to make sure they explicitly deal with invalid json

snuyanzin · 2026-07-17T11:56:46Z

+        if (value instanceof Map) {
+            return ((Map<?, ?>) value).size();
+        }
+        if (value instanceof Collection) {
+            return ((Collection<?>) value).size();
+        }
+        // Scalars, including a JSON null literal, have length 1.
+        return 1;


are we sure it can not be array here?

still actual I guess

snuyanzin · 2026-07-17T17:29:10Z

+            generateCallWithStmtIfArgsNotNull(ctx, resultType, operands, resultNullable = true) {
+              argTerms =>
+                val inputTerm = s"${argTerms.head}.toString()"
+                val parsed = JsonCodeGenHelper.getOrCreateParsedJson(ctx, inputTerm)
+                val varName = parsed.varName
+                val (method, terms) =
+                  if (operands.length > 1)
+                    (BuiltInMethods.JSON_LENGTH_PATH, Seq(varName, s"${argTerms(1)}.toString()"))
+                  else
+                    (BuiltInMethods.JSON_LENGTH, Seq(varName))
+                (parsed.parseCode, s"${qualifyMethod(method)}(${terms.mkString(", ")})")


is there a reason we still have this code here?
Didn't we say that it should be extracted in java class?

snuyanzin · 2026-07-17T20:24:04Z

+        JSON_LENGTH('hello')
+
+        -- 1
+        JSON_LENGTH('{1: "hello", 2: "bye bye"}', '$.2')


the example is invalid

you can check it's output in flink-sql tool

also see https://www.json.org/json-en.html
for more details: only string key is allowed

snuyanzin · 2026-07-17T20:25:35Z

+      Returns NULL if the argument is NULL, the json is invalid, or the path does not locate a value.
+      eg. 
+        -- 2
+        JSON_LENGTH('{1: "hello", 2: "bye bye"}')


it does not work for the case of numbers in keys

moreover: json itself is invalid in this example

snuyanzin

Please fix examples:
in a number of examples json is invalid

snuyanzin · 2026-07-17T20:30:13Z

+      Returns NULL if the argument is NULL, the json is invalid, or the path does not locate a value.
+      eg. 
+        -- 2
+        JSON_LENGTH('{1: "hello", 2: "bye bye"}')


invalid json

snuyanzin

it looks like we still do more parse operations than we need
especially here https://github.com/apache/flink/pull/28688/changes#r3658409889

snuyanzin · 2026-07-27T15:36:08Z

Also this query is still failing (with empty path)

SELECT  json_length('{}', '');

as

Caused by: java.lang.IllegalArgumentException: path can not be null or empty
	at org.apache.flink.shaded.com.jayway.jsonpath.internal.Utils.notEmpty(Utils.java:401)
	at org.apache.flink.shaded.com.jayway.jsonpath.internal.JsonContext.read(JsonContext.java:76)
	at org.apache.flink.table.runtime.functions.SqlJsonUtils.pathExists(SqlJsonUtils.java:453)
	at org.apache.flink.table.runtime.functions.SqlJsonUtils.jsonLength(SqlJsonUtils.java:410)
	at StreamExecCalc$5.processElement(Unknown Source)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollect(StreamSourceContexts.java:389)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collect(StreamSourceContexts.java:484)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$SwitchingOnClose.collect(StreamSourceContexts.java:74)
	at org.apache.flink.streaming.api.functions.source.legacy.InputFormatSourceFunction.run(InputFormatSourceFunction.java:98)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:107)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:68)

snuyanzin · 2026-08-03T05:52:24Z

+    public static Integer jsonLength(final JsonValueContext parsedInput) {
+        // TODO FLINK-40233: A null context can result from a shared parse that was short-circuited
+        // before parsing.
+        if (parsedInput.hasException()) {


parsedInput might be null if input was null e.g. for queries

SELECT json_value(v, cast(null as string)), json_length(v) FROM (values('{"a":1, "b":2}')) AS t(v);

snuyanzin · 2026-08-03T05:52:57Z

@@ -374,6 +389,72 @@ private static Object errorResultForJsonQuery(
        }
    }

+    /** Accepts a pre-parsed context from {@link #jsonParse}. */
+    public static Integer jsonLength(final JsonValueContext parsedInput) {
+        // TODO FLINK-40233: A null context can result from a shared parse that was short-circuited


I guess no need for this TODO anymore since it is merged

…single parse

[FLINK-40089][table] decided how to deal with lax *

… class

…ith JSON_LENGTH + pr issues

…null, true, false' issue

…with jsonlength + to javadoc

…JS_VAL,JS_QUERY,JS_LENGTH at once)

…mpty path

…() + moved jsLeng in sqlFuncZh.yml

raminqaf reviewed Jul 8, 2026

View reviewed changes

gustavodemorais reviewed Jul 8, 2026

View reviewed changes

snuyanzin requested changes Jul 8, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Jul 8, 2026

snuyanzin reviewed Jul 8, 2026

View reviewed changes

snuyanzin requested changes Jul 8, 2026

View reviewed changes

davidradl reviewed Jul 13, 2026

View reviewed changes