raminqaf commented on code in PR #28688:
URL: https://github.com/apache/flink/pull/28688#discussion_r3543949639
##########
flink-python/docs/reference/pyflink.table/expressions.rst:
##########
@@ -319,13 +319,13 @@ JSON functions
.. autosummary::
:toctree: api/
-
Review Comment:
Revert this
##########
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)
+
+
+
+
Review Comment:
Can we reduce the newlines here?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -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.
Review Comment:
Let's use a list here to break down the javadocs
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -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
Review Comment:
```suggestion
* the document as a whole. Returns {@code NULL} if any argument is
{@code NULL} or the path does not
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -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
Review Comment:
remove
```suggestion
```
##########
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
Review Comment:
Remove
```suggestion
```
##########
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.
Review Comment:
Should we consider adding a tiny example?
##########
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.
Review Comment:
How about?
```suggestion
The length is determined as follows:
Scalar values (number, string, boolean, or null): has length 1.
Array: has a length equal to the number of its elements.
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.
```
##########
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) {
Review Comment:
change to Integer and directly return null instead of -1 and rechecking again
```suggestion
private static @Nullable Integer computeLength(JsonNode jsonNode) {
```
##########
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;
Review Comment:
If you you change the return type to `Integer` you can directly return null.
```suggestion
return computeLength(jsonNode);
```
##########
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;
+ }
Review Comment:
Why to we need this? Don't we have already the runtime class
`JsonLengthFunction`?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -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());
Review Comment:
Move these above the comment `// invalid JSON -> NULL` I though these are
all invalid JSON
##########
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) {
Review Comment:
nit: here and everywhere else make parameters/variables that are immutable
`final` (not enforced on Flink but nicer to read).
```suggestion
public @Nullable Integer eval(final @Nullable StringData jsonInput) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
