This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9ee10372283 [FLINK-40390][table] Optimised JSON_LENGTH codegen
9ee10372283 is described below
commit 9ee10372283a09bf19c92738d29a08bdabc08569
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Thu Aug 20 16:43:18 2026 +0100
[FLINK-40390][table] Optimised JSON_LENGTH codegen
This closes #28978.
---
.../functions/BuiltInFunctionDefinitions.java | 29 +++--
.../table/planner/codegen/JsonCodeGenUtils.java | 93 ++++++++++----
.../planner/codegen/calls/JsonLengthCallGen.java | 136 ---------------------
.../table/planner/codegen/ExprCodeGenerator.scala | 2 +-
.../planner/codegen/calls/BuiltInMethods.scala | 3 +-
.../table/planner/codegen/JsonParseReuseTest.java | 12 ++
.../planner/functions/JsonFunctionsITCase.java | 38 ++++--
.../table/runtime/functions/SqlJsonUtils.java | 26 ++--
8 files changed, 138 insertions(+), 201 deletions(-)
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 85edd2ead05..46e0583bb8c 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
@@ -3086,19 +3086,24 @@ public final class BuiltInFunctionDefinitions {
.name("JSON_LENGTH")
.kind(SCALAR)
.inputTypeStrategy(
- or(
-
sequence(logical(LogicalTypeFamily.CHARACTER_STRING)),
- sequence(logical(LogicalTypeRoot.VARIANT)),
- sequence(
-
logical(LogicalTypeFamily.CHARACTER_STRING),
- and(
-
logical(LogicalTypeFamily.CHARACTER_STRING),
- LITERAL)),
- sequence(
- logical(LogicalTypeRoot.VARIANT),
- and(
+ plainJsonPath(
+ or(
+
sequence(logical(LogicalTypeFamily.CHARACTER_STRING)),
+
sequence(logical(LogicalTypeRoot.VARIANT)),
+ sequence(
logical(LogicalTypeFamily.CHARACTER_STRING),
- LITERAL))))
+ and(
+ logical(
+
LogicalTypeFamily
+
.CHARACTER_STRING),
+ LITERAL)),
+ sequence(
+
logical(LogicalTypeRoot.VARIANT),
+ and(
+ logical(
+
LogicalTypeFamily
+
.CHARACTER_STRING),
+ LITERAL)))))
.outputTypeStrategy(explicit(DataTypes.INT().nullable()))
.runtimeProvided()
.build();
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
index 0b466bded2c..816a4325cf8 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/JsonCodeGenUtils.java
@@ -22,6 +22,8 @@ import
org.apache.flink.table.planner.codegen.calls.BuiltInMethods;
import org.apache.flink.table.runtime.functions.SqlJsonUtils;
import org.apache.flink.table.types.logical.LogicalType;
+import java.lang.reflect.Method;
+
import scala.Option;
import scala.Tuple2;
import scala.collection.Seq;
@@ -39,7 +41,6 @@ public final class JsonCodeGenUtils {
*/
public static GeneratedExpression generateJsonType(
CodeGeneratorContext ctx, LogicalType returnType,
Seq<GeneratedExpression> operands) {
- boolean hasPath = operands.length() == 2;
return GenerateUtils.generateCallWithStmtIfArgsNotNull(
ctx,
returnType,
@@ -47,32 +48,76 @@ public final class JsonCodeGenUtils {
true,
false,
argTerms -> {
- ParsedJson parsed = getOrCreateParsedJson(ctx,
argTerms.head() + ".toString()");
- String call;
- if (hasPath) {
- String pathSpec =
operands.apply(1).literalValue().get().toString();
- boolean definite =
SqlJsonUtils.isPathDefinite(pathSpec);
- call =
-
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_TYPE_PATH())
- + "("
- + parsed.varName
- + ", "
- + argTerms.apply(1)
- + ".toString(), "
- + definite
- + ")";
- } else {
- call =
-
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_TYPE())
- + "("
- + parsed.varName
- + ")";
- }
- String resultExpr = CodeGenUtils.BINARY_STRING() +
".fromString(" + call + ")";
- return new Tuple2<>(parsed.parseCode, resultExpr);
+ Tuple2<String, String> parsedCall =
+ generateCallOnParsedInput(
+ ctx,
+ operands,
+ argTerms,
+ BuiltInMethods.JSON_TYPE(),
+ BuiltInMethods.JSON_TYPE_PATH());
+ String resultExpr =
+ CodeGenUtils.BINARY_STRING() + ".fromString(" +
parsedCall._2() + ")";
+ return new Tuple2<>(parsedCall._1(), resultExpr);
});
}
+ /**
+ * Generates {@code JSON_LENGTH(jsonValue)} or {@code
JSON_LENGTH(jsonValue, path)}.
+ *
+ * <p>The parsed context is shared with the other JSON functions over the
same input, so the
+ * input is parsed only once per record.
+ */
+ public static GeneratedExpression generateJsonLength(
+ CodeGeneratorContext ctx, LogicalType returnType,
Seq<GeneratedExpression> operands) {
+ return GenerateUtils.generateCallWithStmtIfArgsNotNull(
+ ctx,
+ returnType,
+ operands,
+ true,
+ false,
+ argTerms ->
+ generateCallOnParsedInput(
+ ctx,
+ operands,
+ argTerms,
+ BuiltInMethods.JSON_LENGTH(),
+ BuiltInMethods.JSON_LENGTH_PATH()));
+ }
+
+ /**
+ * Builds the call against the shared parsed input: the whole-document
overload, or the path
+ * overload with the {@code isPathDefinite} flag resolved from the path
literal at plan time via
+ * {@link SqlJsonUtils#isPathDefinite(String)}.
+ *
+ * @return the parse statement and the call expression
+ */
+ private static Tuple2<String, String> generateCallOnParsedInput(
+ CodeGeneratorContext ctx,
+ Seq<GeneratedExpression> operands,
+ Seq<String> argTerms,
+ Method wholeDocument,
+ Method withPath) {
+ final ParsedJson parsed = getOrCreateParsedJson(ctx, argTerms.head() +
".toString()");
+ if (argTerms.length() == 1) {
+ return new Tuple2<>(
+ parsed.parseCode,
+ CodeGenUtils.qualifyMethod(wholeDocument) + "(" +
parsed.varName + ")");
+ }
+
+ final String pathSpec =
operands.apply(1).literalValue().get().toString();
+ final boolean isPathDefinite = SqlJsonUtils.isPathDefinite(pathSpec);
+ return new Tuple2<>(
+ parsed.parseCode,
+ CodeGenUtils.qualifyMethod(withPath)
+ + "("
+ + parsed.varName
+ + ", "
+ + argTerms.apply(1)
+ + ".toString(), "
+ + isPathDefinite
+ + ")");
+ }
+
/**
* Emits code that parses the given JSON {@code inputTerm} into a reusable
{@link
* SqlJsonUtils.JsonValueContext} member variable.
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/calls/JsonLengthCallGen.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/calls/JsonLengthCallGen.java
deleted file mode 100644
index fffe3e80d3f..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/calls/JsonLengthCallGen.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * 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.planner.codegen.calls;
-
-import org.apache.flink.table.planner.codegen.CodeGenUtils;
-import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
-import org.apache.flink.table.planner.codegen.GeneratedExpression;
-import org.apache.flink.table.runtime.functions.SqlJsonUtils;
-import org.apache.flink.table.types.logical.LogicalType;
-
-import scala.Option;
-import scala.collection.Seq;
-
-/**
- * {@link CallGenerator} for {@code JSON_LENGTH}.
- *
- * <p>The JSON input is parsed into a reusable {@link
SqlJsonUtils.JsonValueContext} that is shared
- * with other JSON functions operating on the same input, so the parse
statement is emitted only
- * once. When a path argument is present the path-aware {@link
BuiltInMethods#JSON_LENGTH_PATH}
- * overload is used, otherwise the whole-document {@link
BuiltInMethods#JSON_LENGTH} overload.
- *
- * <p>The result is nullable: besides propagating a {@code NULL} argument,
{@code JSON_LENGTH}
- * itself returns {@code NULL} for invalid JSON, a path that matches nothing,
or a wildcard path
- * that matches two or more nodes.
- */
-public class JsonLengthCallGen implements CallGenerator {
-
- @Override
- public GeneratedExpression generate(
- CodeGeneratorContext ctx, Seq<GeneratedExpression> operands,
LogicalType returnType) {
-
- String inputTerm = operands.apply(0).resultTerm() + ".toString()";
-
- // Parse the JSON input into a reusable context. When multiple JSON
functions share the
- // same input expression the parse statement is emitted only once and
reused.
- Option<GeneratedExpression> existing =
- ctx.getReusableInputUnboxingExprs(inputTerm,
Integer.MIN_VALUE);
- final String parsedVar;
- final String parseCode;
- if (existing.isDefined()) {
- parsedVar = existing.get().resultTerm();
- parseCode = "";
- } else {
- parsedVar = CodeGenUtils.newName(ctx, "jsonParsed");
- String typeName = SqlJsonUtils.JsonValueContext.class.getName();
- ctx.addReusableMember(typeName + " " + parsedVar + ";");
- ctx.addReusableInputUnboxingExprs(
- inputTerm,
- Integer.MIN_VALUE,
- new GeneratedExpression(parsedVar, "false", "", null,
Option.empty()));
- parseCode =
- parsedVar
- + " = "
- +
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_PARSE())
- + "("
- + inputTerm
- + ");";
- }
-
- final String lengthCall;
- if (operands.length() > 1) {
- lengthCall =
-
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_LENGTH_PATH())
- + "("
- + parsedVar
- + ", "
- + operands.apply(1).resultTerm()
- + ".toString())";
- } else {
- lengthCall =
- CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_LENGTH())
- + "("
- + parsedVar
- + ")";
- }
-
- String resultTypeTerm = CodeGenUtils.boxedTypeTermForType(returnType);
- String defaultValue = CodeGenUtils.primitiveDefaultValue(returnType);
- String nullTerm = ctx.addReusableLocalVariable("boolean", "isNull");
- String resultTerm = ctx.addReusableLocalVariable(resultTypeTerm,
"result");
-
- StringBuilder argsNull = new StringBuilder();
- StringBuilder argsCode = new StringBuilder();
- for (int i = 0; i < operands.length(); i++) {
- GeneratedExpression operand = operands.apply(i);
- if (i > 0) {
- argsNull.append(" || ");
- }
- argsNull.append(operand.nullTerm());
- argsCode.append(operand.code()).append("\n");
- }
-
- String code =
- argsCode
- + nullTerm
- + " = "
- + argsNull
- + ";\n"
- + resultTerm
- + " = "
- + defaultValue
- + ";\n"
- + "if (!"
- + nullTerm
- + ") {\n"
- + parseCode
- + "\n"
- + resultTerm
- + " = "
- + lengthCall
- + ";\n"
- + nullTerm
- + " = ("
- + resultTerm
- + " == null);\n"
- + "}\n";
-
- return new GeneratedExpression(resultTerm, nullTerm, code, returnType,
Option.empty());
- }
-}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
index 7e5779ea8c6..86ffa7b08e9 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
@@ -937,7 +937,7 @@ class ExprCodeGenerator(
new JsonStringCallGen(call, rexProgram).generate(ctx, operands,
resultType)
case BuiltInFunctionDefinitions.JSON_LENGTH =>
- new JsonLengthCallGen().generate(ctx, operands, resultType)
+ JsonCodeGenUtils.generateJsonLength(ctx, resultType, operands)
case BuiltInFunctionDefinitions.JSON_TYPE =>
JsonCodeGenUtils.generateJsonType(ctx, resultType, operands)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
index 99ea6eb483f..d9378d7116f 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -497,7 +497,8 @@ object BuiltInMethods {
classOf[SqlJsonUtils],
"jsonLength",
classOf[SqlJsonUtils.JsonValueContext],
- classOf[String])
+ classOf[String],
+ classOf[Boolean])
val JSON_QUERY = Types.lookupMethod(
classOf[SqlJsonUtils],
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
index abceea8e83c..1e9f4ff689c 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java
@@ -454,4 +454,16 @@ class JsonParseReuseTest {
.as("JSON_LENGTH + JSON_QUERY on the same input should parse
once")
.isOne();
}
+
+ @Test
+ void testJsonLengthAndJsonTypeMixed() {
+ final String sql =
+ "SELECT JSON_LENGTH(json_data, '$.roles'),
JSON_TYPE(json_data, '$.age') "
+ + "FROM json_src";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).containsExactlyInAnyOrder(Row.of(2, "number"),
Row.of(1, "number"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("JSON_LENGTH + JSON_TYPE on the same input should parse
once")
+ .isOne();
+ }
}
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 1301a73254f..3d2e36ba482 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
@@ -120,10 +120,11 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase
{
"$",
"{\"a\":[true, false, null]}",
"{}",
- "[]")
+ "[]",
+ "{\"lax\": {\"strict\": 2}, \"strict value\": 1}")
.andDataTypes(
STRING(), STRING(), STRING(), STRING(), STRING(),
STRING(), STRING(),
- STRING(), STRING(), STRING(), STRING())
+ STRING(), STRING(), STRING(), STRING(), STRING())
// path exists but resolves to a JSON null literal -> scalar,
length 1
.testResult(
$("f8").jsonLength("$.a[2]"),
@@ -313,19 +314,32 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase
{
null,
INT().nullable())
- // lax/strict path modes are not supported and are rejected at
runtime
- .testSqlRuntimeError(
+ // lax/strict path modes are not supported and are rejected at
planning time
+ .testSqlValidationError(
"JSON_LENGTH(f0, 'strict $.type')",
- TableRuntimeException.class,
- "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix (got: 'strict $.type').")
- .testSqlRuntimeError(
+ "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix "
+ + "(got: 'strict $.type'). Use a plain path
such as '$.a.b'. "
+ + "To check path existence or handle invalid
input, use "
+ + "JSON_EXISTS or IS JSON.")
+ .testSqlValidationError(
"JSON_LENGTH(f0, 'lax $.type')",
- TableRuntimeException.class,
- "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix (got: 'lax $.type').")
- .testTableApiRuntimeError(
+ "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix "
+ + "(got: 'lax $.type'). Use a plain path such
as '$.a.b'. "
+ + "To check path existence or handle invalid
input, use "
+ + "JSON_EXISTS or IS JSON.")
+ .testTableApiValidationError(
$("f0").jsonLength("strict $.type"),
- TableRuntimeException.class,
- "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix (got: 'strict $.type').");
+ "JSON_LENGTH does not support the 'lax'/'strict' path
mode prefix "
+ + "(got: 'strict $.type'). Use a plain path
such as '$.a.b'. "
+ + "To check path existence or handle invalid
input, use "
+ + "JSON_EXISTS or IS JSON.")
+ .testResult(
+ $("f11").jsonLength("lax"), "JSON_LENGTH(f11, 'lax')",
1, INT().nullable())
+ .testResult(
+ $("f11").jsonLength("$[\"strict value\"]"),
+ "JSON_LENGTH(f11, '$[\"strict value\"]')",
+ 1,
+ INT().nullable());
}
private static TestSetSpec jsonExistsSpec() {
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 e921cca0324..bcbf7a3f5f2 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
@@ -406,24 +406,20 @@ public class SqlJsonUtils {
return jsonLengthValue(parsedInput.obj);
}
- /** Accepts a pre-parsed context from {@link #jsonParse}. */
- public static Integer jsonLength(final JsonValueContext parsedInput, final
String pathSpec) {
+ /**
+ * Accepts a pre-parsed context from {@link #jsonParse}. {@code
isPathDefinite} is computed at
+ * plan time by {@link #isPathDefinite}.
+ */
+ public static Integer jsonLength(
+ final JsonValueContext parsedInput,
+ final String pathSpec,
+ final boolean isPathDefinite) {
// An empty path is ruled out up front because JsonPath rejects it
with an
// IllegalArgumentException instead of the InvalidPathException caught
below.
if (parsedInput == null || parsedInput.hasException() ||
pathSpec.isEmpty()) {
return null;
}
- final Matcher matcher = JSON_PATH_BASE.matcher(pathSpec);
- final boolean isExplicitLaxStrict = matcher.matches();
- if (isExplicitLaxStrict) {
- throw new TableRuntimeException(
- String.format(
- "JSON_LENGTH does not support the 'lax'/'strict'
path mode prefix (got: '%s'). "
- + "Use a plain path such as '$.a.b'. To
check path existence or handle "
- + "invalid input, use JSON_EXISTS or IS
JSON.",
- pathSpec));
- }
// JsonPath rejects a null root document, so a whole document that is
a JSON null literal
// has to be resolved here. Only the root path matches it, as a scalar
of length 1.
if (parsedInput.obj == null) {
@@ -437,7 +433,7 @@ public class SqlJsonUtils {
return null;
}
- if (!JsonPath.isPathDefinite(pathSpec)) {
+ if (!isPathDefinite) {
final List<?> matched = (List<?>) value;
return matched.size() == 1 ? jsonLengthValue(matched.get(0)) :
null;
}
@@ -567,7 +563,7 @@ public class SqlJsonUtils {
* exactly one value. {@code definite} is computed at plan time by {@link
#isPathDefinite}.
*/
public static String jsonType(
- final JsonValueContext parsedInput, final String path, final
boolean definite) {
+ final JsonValueContext parsedInput, final String path, final
boolean isPathDefinite) {
if (parsedInput == null || parsedInput.hasException() ||
path.isEmpty()) {
return null;
}
@@ -584,7 +580,7 @@ public class SqlJsonUtils {
return null;
}
- if (!definite) {
+ if (!isPathDefinite) {
// Indefinite paths (e.g. wildcards) read back as a list; only one
match has one type.
final List<?> matched = (List<?>) value;
return matched.size() == 1 ? getJsonType(matched.get(0)) : null;
