[FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule by gustavodemorais · Pull Request #27733 · apache/flink

gustavodemorais · 2026-03-04T08:49:29Z

What is the purpose of the change

Add a new planner optimization rule that simplifies COALESCE expressions referencing columns from opposite sides of an equi-join condition, replacing them with a direct column reference to the preserved (non-null) side.

Brief change log

Added SimplifyCoalesceWithEquiJoinConditionRule that rewrites COALESCE(a.col, b.col) to a direct column reference when the columns form an equi-join pair and the join type guarantees one side is non-null
Registered the rule in both FlinkBatchRuleSets and FlinkStreamRuleSets

Verifying this change

Added SimplifyCoalesceWithEquiJoinConditionRuleTest covering LEFT, RIGHT, INNER, and FULL OUTER joins, composite keys, non-equi columns, three-arg COALESCE, and nested expressions

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)

…tionRule

flinkbot · 2026-03-04T08:56:00Z

CI report:

3768a66 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

snuyanzin · 2026-03-04T09:40:32Z

...apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java

+            simplified = true;
+
+            // Handle potential type mismatch by adding a CAST if needed
+            if (call.getType().equals(preservedRef.getType())) {


why LogicalTypeCast#supportsImplicitCast is not an option here?

It's a good option 👍

I think in practice the types in the join conditions are always or almost always the same so the behavior will be the same but I think with implicit casts it's slightly more broad and accurate.

snuyanzin · 2026-03-04T09:41:10Z

...he/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java

+    void testCoalesceOnInnerJoinEquiKey() {
+        util.verifyRelPlan(
+                "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+                        + "FROM orders a INNER JOIN order_details b ON a.order_id = b.order_id");


what about cases with COALESCE in condition?
is it out of scope?

Yeah, it's out of scope and doesn't apply here: when it's in the join condition, we have to evaluate it for all records. So it happens before the join and there's no preserved side or something like that

snuyanzin · 2026-03-04T09:42:31Z

...apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java

+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule


Do we need a separate rule or we can combine it with RemoveUnreachableCoalesceArgumentsRule?

I think having different rules is cleaner since they work differently:

RemoveUnreachableCoalesceArgumentsRule trims arguments based on type nullability metadata (e.g., COALESCE(nonNull, x) -> nonNull), without any join awareness.

This one is specific to joins and merging both would look less elegant IMO.

…k for type equality

snuyanzin · 2026-03-05T11:55:51Z

...he/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java

+    void testNestedCoalesceInExpression() {
+        util.verifyRelPlan(
+                "SELECT CAST(COALESCE(b.order_id, a.order_id) AS STRING) AS order_id_str "
+                        + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+    }


I noticed it is not really about nested...

I tried to played locally with nested like I tweaked order_details in a way

util.tableEnv() .executeSql( "CREATE TABLE order_details (" + " r ROW<order_id BIGINT NOT NULL, order_name STRING NOT NULL> NOT NULL, " + " detail STRING," + " PRIMARY KEY (r) NOT ENFORCED" + ") WITH ('connector' = 'values')");

and then the test

util.verifyRelPlan( "SELECT CAST(COALESCE(b.r, ROW(a.order_id, 'e')) AS STRING) AS order_id_str " + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.r.order_id");

and it shows that COALESCE is still present in optimized rel plan
e.g.

Calc(select=[CAST(COALESCE(r, ROW(order_id, 'e')) AS VARCHAR(2147483647)) AS order_id_str]) +- Join(joinType=[LeftOuterJoin], where=[=(order_id, $f2)], select=[order_id, r, $f2], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[HasUniqueKey]) :- Exchange(distribution=[hash[order_id]]) : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id], metadata=[]]], fields=[order_id]) +- Exchange(distribution=[hash[$f2]]) +- Calc(select=[r, r.order_id AS $f2]) +- TableSourceScan(table=[[default_catalog, default_database, order_details, project=[r], metadata=[]]], fields=[r])

since r is declared as ROW(...) NOT NULL i would expect removal of COALESCE

or did I miss anything?

The name was misleading 👍 The test is actually about being inside the cast. Renamed it to testCoalesceWrappedInCast.

Regarding structured type/row support:

We support structured type field access. Added a new test testCoalesceOnNestedRowScalarField

We don't support the case you mentioned. 'order_name' and the literal 'e' here are not always equal, so we cannot simplify this

Additionally, entire rows as join conditions are out of scope - also not a common/recommended practice afaik

…stCoalesceOnNestedRowScalarField

snuyanzin · 2026-03-05T17:30:01Z

...he/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java

+        util.tableEnv()
+                .executeSql(
+                        "CREATE TABLE order_details_row ("
+                                + "  r ROW<order_id BIGINT NOT NULL, order_name STRING NOT NULL> NOT NULL, "
+                                + "  detail STRING,"
+                                + "  PRIMARY KEY (r) NOT ENFORCED"
+                                + ") WITH ('connector' = 'values')");
+


I guess we should follow same approach and move to the top of test class

snuyanzin · 2026-03-05T17:37:24Z

...he/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java

+        util.verifyRelPlan(
+                "SELECT COALESCE(b.order_id, a.order_id, 0) AS order_id "
+                        + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");


Looks like I found an issue with COALESCE

it works if we have NOT NULL at the first position and don't have NULLs
however it doesn't seem to work if we have NULL at first position
like query

SELECT COALESCE(NULL, a.order_id, 0) AS order_id "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id

doesn't matter how many args > 1
and even more interesting thing: doesn't matter whether null is at the first position or not
for instance this query also doesn't remove coalesce

SELECT COALESCE(b.order_id, NULL, a.order_id, 0) AS order_id "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id

btw if I add NULL at the end it works
like

SELECT COALESCE(b.order_id, a.order_id, NULL) AS order_id "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id

...che/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.xml

gustavodemorais added 2 commits March 3, 2026 22:50

[FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule

9e55db9

[FLINK-39202][table] Plan tests for SimplifyCoalesceWithEquiJoinCondi…

3f3532d

…tionRule

snuyanzin reviewed Mar 4, 2026

View reviewed changes

[FLINK-39202][table] Use LogicalTypeCast#supportsImplicitCast to chec…

d9c23f0

…k for type equality

snuyanzin reviewed Mar 5, 2026

View reviewed changes

[FLINK-39202][table] Rename testNestedCoalesceInExpression and add te…

3768a66

…stCoalesceOnNestedRowScalarField

snuyanzin reviewed Mar 5, 2026

View reviewed changes

...che/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.xml Show resolved Hide resolved

Conversation

gustavodemorais commented Mar 4, 2026

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 Mar 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

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!

snuyanzin Mar 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Mar 4, 2026 •

edited

Loading

snuyanzin Mar 5, 2026 •

edited

Loading