[FLINK-40089][table] Implement JSON_LENGTH function

[VasShabu]

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]

CI report:

• 200b15a 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]

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

[raminqaf]

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]

We have these utilities here and some other in JsonLengthFuntion.java itself. Check if we don't already have utilities for performing the length check. If not, check if it makes sense to move the new private utilities you added to this file

[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

1. 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.
2. 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]

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]

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:

1. 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');
2. 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?
3. 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