[FLINK-40233][table] Correctly share the parsed JSON across JSON_VALUE/JSON_QUERY calls when the owning call is skipped because of NULL or not taken branch

[snuyanzin]

What is the purpose of the change

It might happen that NULL is cached while it should not for instance in case of

SELECT json_value(v, CAST(NULL AS STRING)), json_value(v, '$.a')
FROM (values('{"a":1, "b":2}')) AS t(v);
-- returns (NULL, NULL), expected (NULL, 1)

Brief change log

codegen

Verifying this change

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): ( no)
• 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? (no)
• If yes, how is the feature documented? (not applicable)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[flinkbot]

CI report:

• bd385c6 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gustavodemorais]

Hey @snuyanzin, thanks quickly picking this one up. I've added some suggestions

[gustavodemorais]

Leftover?

Suggested change

	.isOdd();
	.isOne();

[gustavodemorais]

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 for me and dropping the test would also be ok

[gustavodemorais]

All tests use a plain column: could you add a test using a computed column like using TRIM(json_data) as the input to the json functions?

[gustavodemorais]

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

[gustavodemorais]

Could we do this?

Suggested change

	|private $typeName $methodName(Object in) {
	|private $typeName $methodName(${CodeGenUtils.BINARY_STRING} in) {