gustavodemorais commented on code in PR #28836:
URL: https://github.com/apache/flink/pull/28836#discussion_r3675065120
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java:
##########
@@ -149,9 +154,81 @@ void testThreeJsonFunctionCalls() {
.containsExactlyInAnyOrder(
Row.of("account", "42", "{\"city\":\"Munich\"}"),
Row.of("admin", "30", "{\"city\":\"Berlin\"}"));
- assertThat(countJsonParse(extractGeneratedCode(sql)))
+ final String code = extractGeneratedCode(sql);
+ assertThat(countJsonParse(code))
.as("Three JSON function calls on the same input should parse
once")
- .isEqualTo(1);
+ .isOne();
+ }
+
+ @Test
+ void testReuseSurvivesCodeSplitting() {
Review Comment:
Are you sure this is testing code splitting?
extractGeneratedCode reads GeneratedClass.getCode(), which is the original
source; splitting is applied in the constructor into the separate splitCode
field and only that one is compiled (GeneratedClass.java:55-62, :97). The
config only affects the collect(sql) half. I think it needs some tweaks in the
test to test splitting
But to be honest is not the most important thing to test in the PR, so I'd
not say this is a blocker and dropping the test would also be ok
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java:
##########
@@ -133,9 +137,10 @@ void testJsonValueAndJsonQueryMixed() {
.containsExactlyInAnyOrder(
Row.of("account", "{\"city\":\"Munich\"}"),
Row.of("admin", "{\"city\":\"Berlin\"}"));
- assertThat(countJsonParse(extractGeneratedCode(sql)))
+ final String code = extractGeneratedCode(sql);
+ assertThat(countJsonParse(code))
.as("JSON_VALUE + JSON_QUERY on the same input should parse
once")
- .isEqualTo(1);
+ .isOdd();
Review Comment:
Leftover?
```suggestion
.isOne();
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java:
##########
@@ -165,4 +242,38 @@ void testDifferentJsonInputs() {
.as("JSON_VALUE calls on different inputs should parse
separately")
.isEqualTo(2);
}
+
Review Comment:
I think I found one edge case, but it might have an easy solution
Repro
```
-- rows: (1000,'{"n":1}') (2000,'{"n":2}') (3000,'{"n":3}') (9000,'{"n":9}')
SELECT total FROM events MATCH_RECOGNIZE (
ORDER BY rt
MEASURES SUM(CAST(JSON_VALUE(A.f1, '$.n') AS INT)) AS total
AFTER MATCH SKIP PAST LAST ROW
PATTERN (A+ B)
DEFINE A AS A.f0 < 9000, B AS B.f0 >= 9000
)
-- this branch: [3] (= 1+1+1)
-- master: [6] (= 1+2+3)
```
The aggregate loops over every row of the match (MatchCodeGenerator.scala),
but jsonParsed$0 = null; is a per-record statement, emitted once per
processMatch - so rows 2..n reuse row 1's parse. Same cause in batch with
table.exec.operator-fusion-codegen.enabled=true: CalcFusionCodegenSpec never
emits reusePerRecordCode(), so the reset is dropped entirely.
Master is safe because the assignment was unconditional at the call site.
The problem is that correctness now depends on which operator template wraps
the expression.
Possible easy solution: A local (addReusableLocalVariable, plus an explicit
= null init) would fix both - reuseLocalVariableCode() is emitted in the fusion
path and is keyed per method, so it resets inside the MATCH loop too
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/JsonParseReuse.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.{CodeGeneratorContext,
CodeGenUtils, GeneratedExpression}
+import org.apache.flink.table.planner.codegen.CodeGenUtils.qualifyMethod
+import org.apache.flink.table.runtime.functions.SqlJsonUtils
+
+/**
+ * Shares the parsed JSON input between the JSON function calls of a single
generated expression,
+ * see [[JsonValueCallGen]] and [[JsonQueryCallGen]].
+ *
+ * The input is parsed at most once per record, by whichever call runs first,
and the result is held
+ * in a member variable that the following calls on the same input read
instead of parsing again.
+ */
+object JsonParseReuse {
+
+ /**
+ * Returns the expression holding the parsed input. The caller generates its
own call with the
+ * original operands and reads the parsed input from the returned expression.
+ *
+ * The parse is lazy: the result term is a call to a member method that
parses on its first
+ * invocation for the record. `generateCallWithStmtIfArgsNotNull` places a
call under a guard that
+ * requires *all* of its arguments to be non-null, so no single call can be
made the owner of the
+ * parse - in
+ * {{{
+ * SELECT JSON_VALUE(v, CAST(NULL AS STRING)), JSON_QUERY(v, '$.a')
+ * }}}
+ * the NULL path makes the first call short-circuit while the second one
still needs the parse.
+ * Conversely, when no call runs, no parse happens at all.
+ */
+ def parseSharedInput(
+ ctx: CodeGeneratorContext,
+ operands: Seq[GeneratedExpression]): GeneratedExpression = {
+ val input = operands.head
+ val inputTerm = s"${input.resultTerm}.toString()"
+
+ ctx.getReusableInputUnboxingExprs(inputTerm, Int.MinValue) match {
+ case Some(expr) => expr
+ case None =>
+ val varName = CodeGenUtils.newName(ctx, "jsonParsed")
+ val methodName = CodeGenUtils.newName(ctx, "parseJson")
+ val typeName = classOf[SqlJsonUtils.JsonValueContext].getName
+ ctx.addReusableMember(s"$typeName $varName;")
+
+ // null means not parsed yet - jsonParse never returns null for a
non-null input
+ ctx.addReusableMember(
+ s"""
+ |private $typeName $methodName(Object in) {
Review Comment:
Could we do this?
```suggestion
|private $typeName $methodName(${CodeGenUtils.BINARY_STRING}
in) {
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/JsonParseReuseTest.java:
##########
@@ -165,4 +242,38 @@ void testDifferentJsonInputs() {
.as("JSON_VALUE calls on different inputs should parse
separately")
.isEqualTo(2);
}
+
+ @Test
+ void testReuseInOverWindowQueryIsResetPerRow() {
+ // JSON scalars run in a Calc ahead of the OverAggregate; each row
must get its own parse
+ final TableEnvironment bEnv =
TableEnvironment.create(EnvironmentSettings.inBatchMode());
+ bEnv.createTemporaryView(
+ "over_src",
+ bEnv.fromValues(Row.of(1, JSON_ROW1), Row.of(2,
JSON_ROW2)).as("id", "j"));
+ final String sql =
+ "SELECT id, "
+ + "MAX(JSON_VALUE(j, '$.type')) OVER (ORDER BY id ROWS
BETWEEN CURRENT ROW AND CURRENT ROW), "
+ + "MAX(JSON_QUERY(j, '$.address')) OVER (ORDER BY id
ROWS BETWEEN CURRENT ROW AND CURRENT ROW) "
+ + "FROM over_src";
+ final List<Row> rows = new ArrayList<>();
+ bEnv.executeSql(sql).collect().forEachRemaining(rows::add);
+ assertThat(rows)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "account", "{\"city\":\"Munich\"}"),
+ Row.of(2, "admin", "{\"city\":\"Berlin\"}"));
+ }
+
+ @Test
+ void testFilterAndProjectionShareParse() {
Review Comment:
All tests use a plain column: could you add a test using a computed column
like using UPPER(json_data) as the input to the json functions?
--
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]
