[FLINK-37725] Makes Async calcs share correlate split rule with Python #26505

AlanConfluent · 2025-04-24T21:25:22Z

What is the purpose of the change

Moves PythonCorrelateSplitRule to be common as RemoteCorrelateSplitRule so that we can move Async Scalars out of correlate queries. Today an attempt to use an AsyncCalc in a correlate results in a codegen error:

    @Test
    public void testTableFuncWithAsyncCalc() {
        Table t1 = tEnv.fromValues(1, 2).as("f1");
        tEnv.createTemporaryView("t1", t1);
        tEnv.createTemporarySystemFunction("func", new RandomTableFunction());
        tEnv.createTemporarySystemFunction("addTen", new AsyncFuncAdd10());
        final List<Row> results = executeSql("select * FROM t1, LATERAL TABLE(func(addTen(f1)))");
        final List<Row> expectedRows =
                Arrays.asList(
                        Row.of(1, "blah 11"),
                        Row.of(1, "foo 11"),
                        Row.of(2, "blah 12"),
                        Row.of(2, "foo 12"));
        assertThat(results).containsSequence(expectedRows);
    }

Results in error:

Caused by: org.codehaus.commons.compiler.CompileException: Line 99, Column 15: Unknown variable or type "f"
    at org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:13080)
    at org.codehaus.janino.UnitCompiler.getType2(UnitCompiler.java:7230)

Brief change log

(for example:)

The TaskInfo is stored in the blob store on job creation time as a persistent artifact
Deployments RPC transmits only the blob storage reference
TaskManagers retrieve the TaskInfo from the blob cache

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

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)
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 · 2025-04-24T21:30:43Z

CI report:

7203ae8 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Copilot

Pull Request Overview

This PR refactors the correlate split rules for async calcs and Python calls by unifying them under a common RemoteCorrelateSplitRule, thereby addressing codegen errors when using AsyncCalc in correlate queries.

Migrates Python-specific correlate splitting to a common rule with PythonRemoteCalcCallFinder
Updates test cases and rule sets to support asynchronous scalar functions in correlate queries
Introduces minor improvements to RemoteCalcCallFinder and AsyncCalcSplitRule implementations

Reviewed Changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.xml	Added test cases verifying the correlate plan transformation with async calcs.
flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/AsyncCalcITCase.java	Introduced a new integration test for table functions with async calcs.
Other test and rule files	Refactored and unified correlate split rules and updated the rule sets accordingly.
PythonCorrelateSplitRule.java	Migrated legacy Python rule logic to leverage RemoteCorrelateSplitRule.
RemoteCorrelateSplitRule.java, AsyncCalcSplitRule.java	Updated internal rule matching and equality logic to incorporate new RemoteCalcCallFinder implementations.

Copilot · 2025-05-14T10:22:46Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+            return callFinder.isRemoteCall(rexNode) && callFinder.containsNonRemoteCall(rexNode)
+                    || callFinder.isNonRemoteCall(rexNode) && callFinder.containsRemoteCall(rexNode)


[nitpick] Consider adding parentheses around the complex boolean condition involving callFinder checks to clarify operator precedence and improve readability.

Suggested change

return callFinder.isRemoteCall(rexNode) && callFinder.containsNonRemoteCall(rexNode)

|| callFinder.isNonRemoteCall(rexNode) && callFinder.containsRemoteCall(rexNode)

return (callFinder.isRemoteCall(rexNode) && callFinder.containsNonRemoteCall(rexNode))

|| (callFinder.isNonRemoteCall(rexNode) && callFinder.containsRemoteCall(rexNode))

That's a good comment from copilot

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java

Copilot · 2025-05-14T10:22:47Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+                        correlate.getRowType(),
+                        newCorrelate);
+
+        call.transformTo(newTopCalc);


Consider overriding hashCode() in RemoteCorrelateSplitRule to maintain consistency with the custom equals() implementation and ensure correct behavior in hash-based collections.

That's a good comment

dawidwys · 2025-05-14T11:29:31Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+
+    RemoteCorrelateSplitRule(RemoteCalcCallFinder callFinder) {
+        super(
+                operand(FlinkLogicalCorrelate.class, any()),


Since you're modifying the class, can you please replace the usage of deprecated operand and any?

Yes, converted to the new form of rule with the config.

dawidwys · 2025-05-14T11:34:46Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+    }
+
+    private ScalarFunctionSplitter createScalarFunctionSplitter(
+            RexProgram program,


IDE tells me it's always null.

I removed the parameter and put null below. I also put a comment about how the scan should not contain any local references that need resolving, in my understanding. I could potentially make the program optional in ScalarFunctionSplitter and assert it exists before use when seeing a local reference, but that seems unnecessary.

dawidwys · 2025-05-14T11:35:25Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+        }
+
+        FlinkLogicalCorrelate newCorrelate;
+        if (extractedRexNodes.size() > 0) {


Suggested change

if (extractedRexNodes.size() > 0) {

if (!extractedRexNodes.isEmpty()) {

dawidwys · 2025-05-14T11:43:31Z

...src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRule.java

+
+    private static final RemoteCalcCallFinder ASYNC_CALL_FINDER = new AsyncRemoteCalcCallFinder();
+
+    public static final RelOptRule CORRELATE_SPLIT =


Suggested change

public static final RelOptRule CORRELATE_SPLIT =

public static final RelOptRule INSTANCE =

dawidwys · 2025-05-14T11:44:19Z

...rc/main/java/org/apache/flink/table/planner/plan/rules/logical/RemoteCorrelateSplitRule.java

+            return callFinder.isRemoteCall(rexNode) && callFinder.containsNonRemoteCall(rexNode)
+                    || callFinder.isNonRemoteCall(rexNode) && callFinder.containsRemoteCall(rexNode)


That's a good comment from copilot

dawidwys · 2025-05-14T11:48:32Z

...test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.java

+import org.junit.jupiter.api.Test;
+
+/** Test for {@link AsyncCorrelateSplitRule}. */
+public class AsyncCorrelateSplitRuleTest extends TableTestBase {


Can you please fix the compiler warnings.

Yes, done. I believe they are all fixed.

dawidwys · 2025-05-14T11:48:37Z

...test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.java

+/** Test for {@link AsyncCorrelateSplitRule}. */
+public class AsyncCorrelateSplitRuleTest extends TableTestBase {
+
+    private TableTestUtil util = streamTestUtil(TableConfig.getDefault());


This can be final

dawidwys · 2025-05-14T11:48:57Z

...test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.java

+
+    @BeforeEach
+    public void setup() {
+        FlinkChainedProgram programs = new FlinkChainedProgram<BatchOptimizeContext>();


Raw use of parameterized class 'FlinkChainedProgram'

dawidwys · 2025-05-14T11:50:08Z

...test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.java

+                FlinkHepRuleSetProgramBuilder.newBuilder()
+                        .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE())
+                        .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+                        .add(FlinkStreamRuleSets.LOGICAL_REWRITE())


You use a BatchOptimizeContext context with Stream rule set.

I think I may have copied this part of the test -- my mistake. Changed to StreamOptimizeContext.

apache#26505)

AlanConfluent force-pushed the FLINK-37725 branch from d002034 to 3539058 Compare April 24, 2025 21:29

AlanConfluent force-pushed the FLINK-37725 branch 2 times, most recently from ec2d2e0 to 8489b71 Compare May 13, 2025 23:37

dawidwys changed the title ~~Makes Async calcs share correlate split rule with Python~~ [FLINK-37725] Makes Async calcs share correlate split rule with Python May 14, 2025

dawidwys requested a review from Copilot May 14, 2025 10:22

Copilot AI reviewed May 14, 2025

View reviewed changes

dawidwys requested changes May 14, 2025

View reviewed changes

flinkbot added the component=<none> label May 14, 2025

[FLINK-37725] Makes Async calcs share correlate split rule with Python

7203ae8

AlanConfluent force-pushed the FLINK-37725 branch from 8489b71 to 7203ae8 Compare May 15, 2025 01:54

dawidwys approved these changes May 15, 2025

View reviewed changes

dawidwys merged commit 53616d6 into apache:master May 15, 2025

yanand0909 pushed a commit to yanand0909/flink that referenced this pull request Jun 10, 2025

[FLINK-37725] Makes Async calcs share correlate split rule with Python (

7e49e1e

apache#26505)

		return callFinder.isRemoteCall(rexNode) && callFinder.containsNonRemoteCall(rexNode)
		\|\| callFinder.isNonRemoteCall(rexNode) && callFinder.containsRemoteCall(rexNode)

	if (extractedRexNodes.size() > 0) {
	if (!extractedRexNodes.isEmpty()) {


		private static final RemoteCalcCallFinder ASYNC_CALL_FINDER = new AsyncRemoteCalcCallFinder();

		public static final RelOptRule CORRELATE_SPLIT =

	public static final RelOptRule CORRELATE_SPLIT =
	public static final RelOptRule INSTANCE =

[FLINK-37725] Makes Async calcs share correlate split rule with Python #26505

[FLINK-37725] Makes Async calcs share correlate split rule with Python #26505

Uh oh!

Conversation

AlanConfluent commented Apr 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Apr 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI May 14, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI May 14, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

AlanConfluent commented Apr 24, 2025 •

edited

Loading

flinkbot commented Apr 24, 2025 •

edited

Loading