[FLINK-37721] Fixes janino bug returning incorrect results

[AlanConfluent]

What is the purpose of the change

The bug comes from the fact that previously, we were creating an anonymous function callback within the janino generated code. This function was then referencing the local DelegatingAsyncResultFuture, effectively creating a closure. When the function was called back from different threads, they seemed to both reference the same DelegatingAsyncResultFuture rather than their own local ones. The result was returning incorrect "synchronous" results, so queries like the following might have issues:

SELECT f1, func(f1, f2, f3) FROM Table1;

Here, the f1 column returned was retrieved by asking for it from the local DelegatingAsyncResultFuture, and so getting the wrong instance meant getting the wrong result.

To fix it, I removed the inner anonymous function and just added the same logic on DelegatingAsyncResultFuture. It requires setting it up with all of the metadata (indexes, sync results), and when the result is complete, rather than calling the generated function, it just creates the result RowData itself.

Brief change log

• Changes how codegen works for async calcs to avoid anonymous functions

Verifying this change

This change added tests and can be verified as follows:

• Added ITCase that previously gave incorrect result, but which now gives the right result.
• Added test which shows that changelog type is propagated

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
  • The former generated code and static code in DelegatingAsyncResultFuture shouldn't differ in any significant way, so should be a no-op.
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[flinkbot]

CI report:

• f8339e2 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[dawidwys]

Could you please fix the PR to meet the community standards? In particular:

1. Please add jira number to each commit and the PR title.
2. Please fill in the PR template

You can find contribution guidelines here: https://flink.apache.org/how-to-contribute/code-style-and-quality-pull-requests/

Thank you

[dawidwys]

The PR should be good after fixing the PR requirements

[dawidwys]

Could you please link the JIRA issue?

[AlanConfluent]

Sounds good, added the link

[dawidwys]

Could you please double check you committed the change. I don't see a JIRA number mentioned in the comment.

[AlanConfluent]

My mistake. I included the link this time.