This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 ce631e0ec6d [FLINK-40901][table] Use `JsonParseReuse` for
`JSON_TYPE`/`JSON_LENGTH`
ce631e0ec6d is described below
commit ce631e0ec6d2b0884b2c835bbe767430d1c033fe
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Tue Oct 13 07:36:53 2026 +0100
[FLINK-40901][table] Use `JsonParseReuse` for `JSON_TYPE`/`JSON_LENGTH`
---
.../table/planner/codegen/JsonCodeGenUtils.java | 85 ++++------------------
.../table/planner/codegen/JsonParseReuseTest.java | 49 ++++++++++++-
2 files changed, 62 insertions(+), 72 deletions(-)
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 816a4325cf8..f63c35085ea 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
@@ -19,13 +19,12 @@
package org.apache.flink.table.planner.codegen;
import org.apache.flink.table.planner.codegen.calls.BuiltInMethods;
+import org.apache.flink.table.planner.codegen.calls.JsonParseReuse;
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;
/** Utilities for the code generation of JSON functions. */
@@ -41,23 +40,21 @@ public final class JsonCodeGenUtils {
*/
public static GeneratedExpression generateJsonType(
CodeGeneratorContext ctx, LogicalType returnType,
Seq<GeneratedExpression> operands) {
- return GenerateUtils.generateCallWithStmtIfArgsNotNull(
+ return GenerateUtils.generateCallIfArgsNotNull(
ctx,
returnType,
operands,
true,
false,
argTerms -> {
- Tuple2<String, String> parsedCall =
+ final String call =
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);
+ return CodeGenUtils.BINARY_STRING() + ".fromString(" +
call + ")";
});
}
@@ -69,7 +66,7 @@ public final class JsonCodeGenUtils {
*/
public static GeneratedExpression generateJsonLength(
CodeGeneratorContext ctx, LogicalType returnType,
Seq<GeneratedExpression> operands) {
- return GenerateUtils.generateCallWithStmtIfArgsNotNull(
+ return GenerateUtils.generateCallIfArgsNotNull(
ctx,
returnType,
operands,
@@ -88,77 +85,27 @@ public final class JsonCodeGenUtils {
* 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(
+ private static String generateCallOnParsedInput(
CodeGeneratorContext ctx,
Seq<GeneratedExpression> operands,
Seq<String> argTerms,
Method wholeDocument,
Method withPath) {
- final ParsedJson parsed = getOrCreateParsedJson(ctx, argTerms.head() +
".toString()");
+ final String parsed = JsonParseReuse.parseSharedInput(ctx,
operands).resultTerm();
if (argTerms.length() == 1) {
- return new Tuple2<>(
- parsed.parseCode,
- CodeGenUtils.qualifyMethod(wholeDocument) + "(" +
parsed.varName + ")");
+ return CodeGenUtils.qualifyMethod(wholeDocument) + "(" + parsed +
")";
}
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.
- *
- * @return the parsed-context variable name and the parse statement, empty
if the same input was
- * already parsed
- */
- private static ParsedJson getOrCreateParsedJson(CodeGeneratorContext ctx,
String inputTerm) {
- Option<GeneratedExpression> existing =
- ctx.getReusableInputUnboxingExprs(inputTerm,
Integer.MIN_VALUE);
- if (existing.isDefined()) {
- return new ParsedJson(existing.get().resultTerm(), "");
- }
-
- String varName = CodeGenUtils.newName(ctx, "jsonParsed");
- String typeName = SqlJsonUtils.JsonValueContext.class.getName();
- ctx.addReusableMember(typeName + " " + varName + ";");
-
- ctx.addReusableInputUnboxingExprs(
- inputTerm,
- Integer.MIN_VALUE,
- new GeneratedExpression(varName, "false", "", null,
Option.empty()));
-
- String parseCode =
- varName
- + " = "
- +
CodeGenUtils.qualifyMethod(BuiltInMethods.JSON_PARSE())
- + "("
- + inputTerm
- + ");";
- return new ParsedJson(varName, parseCode);
- }
-
- /** Holds the outcome of {@link #getOrCreateParsedJson}. */
- private static final class ParsedJson {
- private final String varName;
- private final String parseCode;
-
- ParsedJson(String varName, String parseCode) {
- this.varName = varName;
- this.parseCode = parseCode;
- }
+ return CodeGenUtils.qualifyMethod(withPath)
+ + "("
+ + parsed
+ + ", "
+ + argTerms.apply(1)
+ + ".toString(), "
+ + isPathDefinite
+ + ")";
}
}
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 1e9f4ff689c..e5dab9dd86d 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
@@ -210,12 +210,14 @@ class JsonParseReuseTest {
// calls on the same computed input (TRIM) must still share a parse
final String sql =
"SELECT JSON_VALUE(TRIM(json_data), '$.type'), "
- + "JSON_QUERY(TRIM(json_data), '$.address') FROM
json_src";
+ + "JSON_QUERY(TRIM(json_data), '$.address'), "
+ + "JSON_TYPE(TRIM(json_data)), "
+ + "JSON_LENGTH(TRIM(json_data)) FROM json_src";
final List<Row> rows = collect(sql);
assertThat(rows)
.containsExactlyInAnyOrder(
- Row.of("account", "{\"city\":\"Munich\"}"),
- Row.of("admin", "{\"city\":\"Berlin\"}"));
+ Row.of("account", "{\"city\":\"Munich\"}", "object",
4),
+ Row.of("admin", "{\"city\":\"Berlin\"}", "object", 4));
final String code = extractGeneratedCode(sql);
assertThat(countJsonParse(code))
.as("Calls on the same computed input should parse once")
@@ -466,4 +468,45 @@ class JsonParseReuseTest {
.as("JSON_LENGTH + JSON_TYPE on the same input should parse
once")
.isOne();
}
+
+ @Test
+ void testJsonLengthAfterSkippedJsonTypeIsResetPerRow() {
+ final String sql =
+ "SELECT id, CASE WHEN id > 1 THEN JSON_TYPE(j) END,
JSON_LENGTH(j) "
+ + "FROM (VALUES (2, '[1,2,3]'), (1, '[1]')) AS t(id,
j)";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).containsExactlyInAnyOrder(Row.of(2, "array", 3),
Row.of(1, null, 1));
+ }
+
+ @Test
+ void testJsonTypeAfterSkippedJsonLengthIsResetPerRow() {
+ final String sql =
+ "SELECT id, CASE WHEN id > 1 THEN JSON_LENGTH(j) END,
JSON_TYPE(j) "
+ + "FROM (VALUES (2, '[1,2,3]'), (1, '{\"a\":1}')) AS
t(id, j)";
+ final List<Row> rows = collect(sql);
+ assertThat(rows)
+ .containsExactlyInAnyOrder(Row.of(2, 3, "array"), Row.of(1,
null, "object"));
+ }
+
+ @Test
+ void testJsonLengthAfterShortCircuitedJsonTypeInFilterIsResetPerRow() {
+ final String sql =
+ "SELECT id, JSON_LENGTH(j) "
+ + "FROM (VALUES (2, '[1,2,3]'), (1, '[1]')) AS t(id,
j) "
+ + "WHERE id > 1 OR JSON_TYPE(j) = 'array'";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).containsExactlyInAnyOrder(Row.of(2, 3), Row.of(1, 1));
+ }
+
+ @Test
+ void testJsonValueAfterSkippedJsonTypeIsResetPerRow() {
+ final String sql =
+ "SELECT id, CASE WHEN id > 1 THEN JSON_TYPE(j) END,
JSON_VALUE(j, '$.a') "
+ + "FROM (VALUES (2, '{\"a\":\"x\"}'), (1,
'{\"a\":\"y\"}')) AS t(id, j)";
+ final List<Row> rows = collect(sql);
+ assertThat(rows).containsExactlyInAnyOrder(Row.of(2, "object", "x"),
Row.of(1, null, "y"));
+ assertThat(countJsonParse(extractGeneratedCode(sql)))
+ .as("JSON_TYPE + JSON_VALUE on the same input should parse
once")
+ .isOne();
+ }
}
