You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gustavodemorais (via GitHub)" <gi...@apache.org> on 2026/03/04 08:49:34 UTC
[PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
gustavodemorais opened a new pull request, #27733:
URL: https://github.com/apache/flink/pull/27733
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## 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)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2913942139
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
Hey @snuyanzin, this is a good catch. This is more of a new optimization that wasn't part of the scope here, since it's not related to joins or columns being equal. (also hope users do not have NULLs in their coalesce calls since that makes little sense)
That said, I agree we can remove NULLs from coalesce but the best place would be the other rule we have `RemoveUnreachableCoalesceArgumentsRule`. I actually drafted the code locally already but think it makes sense to do it in another PR since it's not related. Wdyt?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2885013419
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
Review Comment:
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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2882792323
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
Review Comment:
Do we need a separate rule or we can combine it with `RemoveUnreachableCoalesceArgumentsRule`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891476259
##########
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.xml:
##########
@@ -0,0 +1,223 @@
+<?xml version="1.0" ?>
+<!--
+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.
+-->
+<Root>
+ <TestCase name="testCoalesceOnFullJoinNotSimplified">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.order_id, a.order_id) AS order_id FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[full])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[COALESCE(order_id0, order_id) AS order_id])
++- Join(joinType=[FullOuterJoin], where=[=(order_id, order_id0)], select=[order_id, order_id0], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id], metadata=[]]], fields=[order_id])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnInnerJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[order_id0 AS order_id])
++- Join(joinType=[InnerJoin], where=[=(order_id, order_id0)], select=[order_id, order_id0], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id], metadata=[]]], fields=[order_id])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnLeftJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)], amount=[$2])
++- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id, amount], metadata=[]]], fields=[order_id, amount])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnNestedRowScalarField">
+ <Resource name="sql">
+ <![CDATA[SELECT CAST(COALESCE(b.r.order_id, a.order_id) AS STRING) AS order_id_str FROM orders a LEFT JOIN order_details_row b ON a.order_id = b.r.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id_str=[CAST(COALESCE($3.order_id, $0)):VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL])
++- LogicalProject(order_id=[$0], user_id=[$1], amount=[$2], r=[$3], detail=[$4])
+ +- LogicalJoin(condition=[=($0, $5)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalProject(r=[$0], detail=[$1], $f2=[$0.order_id])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details_row]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[CAST(r.order_id 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_row, project=[r], metadata=[]]], fields=[r])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnNonEquiColumnsNotSimplified">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(val=[COALESCE($4, CAST($2):VARCHAR(2147483647) CHARACTER SET "UTF-16LE")])
++- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[COALESCE(detail, CAST(amount AS VARCHAR(2147483647))) AS val])
++- Join(joinType=[LeftOuterJoin], where=[=(order_id, order_id0)], select=[order_id, amount, order_id0, detail], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id, amount], metadata=[]]], fields=[order_id, amount])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details]], fields=[order_id, detail])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnRightJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(a.order_id, b.order_id) AS order_id FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($0, $3)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[right])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
Review Comment:
so we read only from `order_details`, how then data could be filtered by this join condition `a.order_id = b.order_id` ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891353453
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
Looks like I found another issue with `COALESCE`
it works if we have `NOT NULL` at the first position
however it doesn't seem to work if we have `NULL` at first position
like query
```sql
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`
```sql
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
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2922936996
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
Review Comment:
then further rules could optimize it
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27733:
URL: https://github.com/apache/flink/pull/27733#issuecomment-3996152119
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3f3532dba0f8f561e16904b3328fd14aed700057",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3f3532dba0f8f561e16904b3328fd14aed700057",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3f3532dba0f8f561e16904b3328fd14aed700057 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2889561185
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
or did I miss anything?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2882782986
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
+
+ final RexNode op0 = call.getOperands().get(0);
+ final RexNode op1 = call.getOperands().get(1);
+ if (!(op0 instanceof RexInputRef) || !(op1 instanceof RexInputRef)) {
+ return call;
+ }
+
+ final RexInputRef ref0 = (RexInputRef) op0;
+ final RexInputRef ref1 = (RexInputRef) op1;
+
+ // Must be from opposite sides of the join
+ final boolean isLeft0 = ref0.getIndex() < leftFieldCount;
+ final boolean isLeft1 = ref1.getIndex() < leftFieldCount;
+ if (isLeft0 == isLeft1) {
+ return call;
+ }
+
+ // Identify left-side and right-side refs
+ final int leftIdx = isLeft0 ? ref0.getIndex() : ref1.getIndex();
+ final int rightIdx = (isLeft0 ? ref1.getIndex() : ref0.getIndex()) - leftFieldCount;
+ final RexInputRef leftRef = isLeft0 ? ref0 : ref1;
+ final RexInputRef rightRef = isLeft0 ? ref1 : ref0;
+
+ // Check if they form an equi-join pair
+ if (!isEquiJoinPair(leftIdx, rightIdx)) {
+ return call;
+ }
+
+ // Determine the preserved-side reference
+ final RexInputRef preservedRef = resolvePreservedSide(leftRef, rightRef, ref0);
+ if (preservedRef == null) {
+ return call;
+ }
+
+ simplified = true;
+
+ // Handle potential type mismatch by adding a CAST if needed
+ if (call.getType().equals(preservedRef.getType())) {
Review Comment:
why `LogicalTypeCast#supportsImplicitCast` is not an option here?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2882786006
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
Review Comment:
what about cases with `COALESCE` in condition?
is it out of scope?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #27733:
URL: https://github.com/apache/flink/pull/27733#issuecomment-4064136286
@gustavodemorais it seems ci is still unhappy, can you take a look please?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891403677
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
btw if I add `NULL` at the end it works
like
```sql
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
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891236859
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
this I didn't get
we know that
```
r is declared as ROW(...)
```
is `NOT NULL` , right?
why it is not enough to replace `COALESCE(b.r, ROW(a.order_id, 'e'))` with `b.r`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2885039730
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
+
+ final RexNode op0 = call.getOperands().get(0);
+ final RexNode op1 = call.getOperands().get(1);
+ if (!(op0 instanceof RexInputRef) || !(op1 instanceof RexInputRef)) {
+ return call;
+ }
+
+ final RexInputRef ref0 = (RexInputRef) op0;
+ final RexInputRef ref1 = (RexInputRef) op1;
+
+ // Must be from opposite sides of the join
+ final boolean isLeft0 = ref0.getIndex() < leftFieldCount;
+ final boolean isLeft1 = ref1.getIndex() < leftFieldCount;
+ if (isLeft0 == isLeft1) {
+ return call;
+ }
+
+ // Identify left-side and right-side refs
+ final int leftIdx = isLeft0 ? ref0.getIndex() : ref1.getIndex();
+ final int rightIdx = (isLeft0 ? ref1.getIndex() : ref0.getIndex()) - leftFieldCount;
+ final RexInputRef leftRef = isLeft0 ? ref0 : ref1;
+ final RexInputRef rightRef = isLeft0 ? ref1 : ref0;
+
+ // Check if they form an equi-join pair
+ if (!isEquiJoinPair(leftIdx, rightIdx)) {
+ return call;
+ }
+
+ // Determine the preserved-side reference
+ final RexInputRef preservedRef = resolvePreservedSide(leftRef, rightRef, ref0);
+ if (preservedRef == null) {
+ return call;
+ }
+
+ simplified = true;
+
+ // Handle potential type mismatch by adding a CAST if needed
+ if (call.getType().equals(preservedRef.getType())) {
Review Comment:
done
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2922933912
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
Review Comment:
can we make a bit more generic?
like if it is 2 then apply the logic like in this rule
if it is > 2 then `COALESCE(..., a.k, ..., b.k, ...)` replace with `COALESCE(..., a.k, ...)` or `COALESCE(..., b.k, ...)`
I still think that this should be done in this rule since this rule has knowledge how to detect such pairs
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2889534064
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
I noticed it is not really about nested...
I tried to played locally with nested like I tweaked `order_details` in a way
```java
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
```java
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])
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891236859
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
this I didn't get
we know that
```
r ROW<order_id BIGINT NOT NULL, order_name STRING NOT NULL> NOT NULL
```
is `NOT NULL` , right?
why it is not enough to replace `COALESCE(b.r, ROW(a.order_id, 'e'))` with `b.r`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2913942139
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
Hey @snuyanzin, this is a good catch. This is more of a new optimization that wasn't part of the scope here, since it's not related to joins or columns being equal. (also hope users do not often have NULLs in their coalesce calls)
That said, I agree we can remove NULLs from coalesce but the best place would be the other rule we have `RemoveUnreachableCoalesceArgumentsRule`. I actually drafted the code locally already but think it makes sense to do it in another PR since it's not related. Wdyt?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #27733:
URL: https://github.com/apache/flink/pull/27733#issuecomment-4067982468
Looks healthy now 🙂
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2917647218
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ void testCoalesceWrappedInCast() {
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnNestedRowScalarField() {
+ 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')");
+
Review Comment:
done
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #27733:
URL: https://github.com/apache/flink/pull/27733
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891327030
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ void testCoalesceWrappedInCast() {
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnNestedRowScalarField() {
+ 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')");
+
Review Comment:
I guess we should follow same approach and move to the top of test class
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891236859
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
this I didn't get
we know that
```
r ROW<order_id BIGINT NOT NULL, order_name STRING NOT NULL> NOT NULL
```
is `NOT NULL` , right?
why it is not enough to replace `COALESCE(b.r, ROW(a.order_id, 'e'))` with `b.r`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2889559853
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
since `r` is declared as `ROW(...) NOT NULL` i would expect removal of `COALESCE`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891158317
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
The name was misleading 👍 The test is actually about being inside the cast. Renamed it to `testCoalesceWrappedInCast`.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891497920
##########
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.xml:
##########
@@ -0,0 +1,223 @@
+<?xml version="1.0" ?>
+<!--
+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.
+-->
+<Root>
+ <TestCase name="testCoalesceOnFullJoinNotSimplified">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.order_id, a.order_id) AS order_id FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[full])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[COALESCE(order_id0, order_id) AS order_id])
++- Join(joinType=[FullOuterJoin], where=[=(order_id, order_id0)], select=[order_id, order_id0], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id], metadata=[]]], fields=[order_id])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnInnerJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[order_id0 AS order_id])
++- Join(joinType=[InnerJoin], where=[=(order_id, order_id0)], select=[order_id, order_id0], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id], metadata=[]]], fields=[order_id])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnLeftJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($3, $0)], amount=[$2])
++- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id, amount], metadata=[]]], fields=[order_id, amount])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnNestedRowScalarField">
+ <Resource name="sql">
+ <![CDATA[SELECT CAST(COALESCE(b.r.order_id, a.order_id) AS STRING) AS order_id_str FROM orders a LEFT JOIN order_details_row b ON a.order_id = b.r.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id_str=[CAST(COALESCE($3.order_id, $0)):VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL])
++- LogicalProject(order_id=[$0], user_id=[$1], amount=[$2], r=[$3], detail=[$4])
+ +- LogicalJoin(condition=[=($0, $5)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalProject(r=[$0], detail=[$1], $f2=[$0.order_id])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details_row]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[CAST(r.order_id 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_row, project=[r], metadata=[]]], fields=[r])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnNonEquiColumnsNotSimplified">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(val=[COALESCE($4, CAST($2):VARCHAR(2147483647) CHARACTER SET "UTF-16LE")])
++- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[COALESCE(detail, CAST(amount AS VARCHAR(2147483647))) AS val])
++- Join(joinType=[LeftOuterJoin], where=[=(order_id, order_id0)], select=[order_id, amount, order_id0, detail], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[order_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, orders, project=[order_id, amount], metadata=[]]], fields=[order_id, amount])
+ +- Exchange(distribution=[hash[order_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, order_details]], fields=[order_id, detail])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCoalesceOnRightJoinEquiKey">
+ <Resource name="sql">
+ <![CDATA[SELECT COALESCE(a.order_id, b.order_id) AS order_id FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[COALESCE($0, $3)])
++- LogicalJoin(condition=[=($0, $3)], joinType=[right])
+ :- LogicalTableScan(table=[[default_catalog, default_database, orders]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, order_details]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, order_details, project=[order_id], metadata=[]]], fields=[order_id])
Review Comment:
ok, it is just left join, then should be ok
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891236859
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
+ }
+
+ @Test
+ void testMultipleCoalesceOnCompositeKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.k1, a.k1) AS k1, COALESCE(b.k2, a.k2) AS k2, a.val "
+ + "FROM composite_key_table a "
+ + "LEFT JOIN composite_key_details b ON a.k1 = b.k1 AND a.k2 = b.k2");
+ }
+
+ @Test
+ 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");
+ }
Review Comment:
this I didn't get
we know that
```
r is declared as ROW(...)
```
is `NOT NULL` , right?
why it is not enough to replace `COALESCE(b.r, ROW(a.order_id, 'e'))` with `b.r`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891353453
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
Looks like I found an issue with `COALESCE`
it works if we have `NOT NULL` at the first position and don't have `NULL`s
however it doesn't seem to work if we have `NULL` at first position
like query
```sql
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`
```sql
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
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2891353453
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRuleTest.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Test rule {@link SimplifyCoalesceWithEquiJoinConditionRule}. */
+class SimplifyCoalesceWithEquiJoinConditionRuleTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE orders ("
+ + " order_id BIGINT NOT NULL,"
+ + " user_id BIGINT NOT NULL,"
+ + " amount DOUBLE,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE order_details ("
+ + " order_id BIGINT NOT NULL,"
+ + " detail STRING,"
+ + " PRIMARY KEY (order_id) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_table ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " val STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE composite_key_details ("
+ + " k1 BIGINT NOT NULL,"
+ + " k2 BIGINT NOT NULL,"
+ + " info STRING,"
+ + " PRIMARY KEY (k1, k2) NOT ENFORCED"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ @Test
+ void testCoalesceOnLeftJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceReversedArgsOnLeftJoin() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id, a.amount "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ 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");
+ }
+
+ @Test
+ void testCoalesceOnRightJoinEquiKey() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(a.order_id, b.order_id) AS order_id "
+ + "FROM orders a RIGHT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnFullJoinNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.order_id, a.order_id) AS order_id "
+ + "FROM orders a FULL JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceOnNonEquiColumnsNotSimplified() {
+ util.verifyRelPlan(
+ "SELECT COALESCE(b.detail, CAST(a.amount AS STRING)) AS val "
+ + "FROM orders a LEFT JOIN order_details b ON a.order_id = b.order_id");
+ }
+
+ @Test
+ void testCoalesceWithThreeArgs() {
+ 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");
Review Comment:
Looks like I found an issue with `COALESCE`
it works if we have `NOT NULL` at the first position
however it doesn't seem to work if we have `NULL` at first position
like query
```sql
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`
```sql
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
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2924141188
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
Review Comment:
Hey Sergey, I looked into it and we can make it a bit more generic in some cases 👍. Look:
Given `a.k` as the preserved side and `b.k` as the nullable side:
Supported:
1. Adjacent before preserved: COALESCE(..., b.k, a.k, ...) -> COALESCE(..., a.k, ...) - when matched b.k=a.k so next arg gives same value; when unmatched b.k is NULL and skipped
2. After preserved: COALESCE(..., a.k, ..., b.k, ...) -> COALESCE(..., a.k, ...) - when matched a.k is non-null so b.k never reached; when unmatched b.k is NULL anyway
Not supported:
1. NOT safe: COALESCE(b.k, x, a.k) - removing b.k changes result when x is non-null (matched returns b.k=a.k, simplified returns x)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2924155851
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
Review Comment:
Done here https://github.com/apache/flink/pull/27733/commits/598d2cd1134eb95e300f3c416dec9fa5f2e24b09
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2885004627
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
+ extends RelRule<SimplifyCoalesceWithEquiJoinConditionRule.Config> {
+
+ public static final RelRule<Config> PROJECT_INSTANCE = Config.DEFAULT.withProject().toRule();
+
+ public static final RelRule<Config> CALC_INSTANCE = Config.DEFAULT.withCalc().toRule();
+
+ public SimplifyCoalesceWithEquiJoinConditionRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final RelNode relNode = call.rel(0);
+ final Join join = call.rel(1);
+
+ final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.pairs().isEmpty()) {
+ return;
+ }
+
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ final int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+
+ final EquiJoinCoalesceSimplifier shuttle =
+ new EquiJoinCoalesceSimplifier(
+ rexBuilder, joinInfo, join.getJoinType(), leftFieldCount);
+
+ final RelNode transformed = relNode.accept(shuttle);
+ if (shuttle.isSimplified()) {
+ call.transformTo(transformed);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A {@link RexShuttle} that replaces {@code COALESCE(refA, refB)} with the preserved-side
+ * column reference when the two arguments form an equi-join pair from opposite sides.
+ */
+ private static class EquiJoinCoalesceSimplifier extends RexShuttle {
+
+ private final RexBuilder rexBuilder;
+ private final JoinInfo joinInfo;
+ private final JoinRelType joinType;
+ private final int leftFieldCount;
+ private boolean simplified = false;
+
+ private EquiJoinCoalesceSimplifier(
+ RexBuilder rexBuilder,
+ JoinInfo joinInfo,
+ JoinRelType joinType,
+ int leftFieldCount) {
+ this.rexBuilder = rexBuilder;
+ this.joinInfo = joinInfo;
+ this.joinType = joinType;
+ this.leftFieldCount = leftFieldCount;
+ }
+
+ boolean isSimplified() {
+ return simplified;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Recurse first (bottom-up) so nested COALESCE calls are handled
+ call = (RexCall) super.visitCall(call);
+
+ if (!operatorIsCoalesce(call.getOperator())) {
+ return call;
+ }
+ if (call.getOperands().size() != 2) {
+ return call;
+ }
+
+ final RexNode op0 = call.getOperands().get(0);
+ final RexNode op1 = call.getOperands().get(1);
+ if (!(op0 instanceof RexInputRef) || !(op1 instanceof RexInputRef)) {
+ return call;
+ }
+
+ final RexInputRef ref0 = (RexInputRef) op0;
+ final RexInputRef ref1 = (RexInputRef) op1;
+
+ // Must be from opposite sides of the join
+ final boolean isLeft0 = ref0.getIndex() < leftFieldCount;
+ final boolean isLeft1 = ref1.getIndex() < leftFieldCount;
+ if (isLeft0 == isLeft1) {
+ return call;
+ }
+
+ // Identify left-side and right-side refs
+ final int leftIdx = isLeft0 ? ref0.getIndex() : ref1.getIndex();
+ final int rightIdx = (isLeft0 ? ref1.getIndex() : ref0.getIndex()) - leftFieldCount;
+ final RexInputRef leftRef = isLeft0 ? ref0 : ref1;
+ final RexInputRef rightRef = isLeft0 ? ref1 : ref0;
+
+ // Check if they form an equi-join pair
+ if (!isEquiJoinPair(leftIdx, rightIdx)) {
+ return call;
+ }
+
+ // Determine the preserved-side reference
+ final RexInputRef preservedRef = resolvePreservedSide(leftRef, rightRef, ref0);
+ if (preservedRef == null) {
+ return call;
+ }
+
+ simplified = true;
+
+ // Handle potential type mismatch by adding a CAST if needed
+ if (call.getType().equals(preservedRef.getType())) {
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39202][table] Add new SimplifyCoalesceWithEquiJoinConditionRule [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27733:
URL: https://github.com/apache/flink/pull/27733#discussion_r2885015444
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SimplifyCoalesceWithEquiJoinConditionRule.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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.plan.rules.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.mapping.IntPair;
+import org.immutables.value.Value;
+
+import java.util.function.Predicate;
+
+/**
+ * Simplifies {@code COALESCE(b.k, a.k)} to the preserved-side column reference when the two
+ * arguments reference columns from opposite sides of an equi-join condition.
+ *
+ * <p>For a {@code LEFT JOIN a ON a.k = b.k}, {@code COALESCE(b.k, a.k)} always equals {@code a.k}:
+ *
+ * <ul>
+ * <li>When b matches: {@code b.k = a.k} (equi-join guarantees this), so the result is {@code
+ * a.k}.
+ * <li>When b doesn't match: {@code b.k} is NULL, so the result falls through to {@code a.k}.
+ * </ul>
+ *
+ * <p>Both orderings ({@code COALESCE(b.k, a.k)} and {@code COALESCE(a.k, b.k)}) resolve to the
+ * preserved-side key. This applies to LEFT, RIGHT, and INNER joins. FULL OUTER joins are not
+ * handled because both sides can generate nulls.
+ *
+ * <p>This rule matches a {@link Project} or {@link Calc} whose input is a {@link Join}. It uses a
+ * {@link RexShuttle} to recursively find and simplify applicable COALESCE calls, including those
+ * nested inside other expressions (e.g., {@code CAST(COALESCE(b.k, a.k) AS VARCHAR)}).
+ */
+@Internal
+@Value.Enclosing
+public class SimplifyCoalesceWithEquiJoinConditionRule
Review Comment:
I think having different rules is cleaner since they work differently:
1. RemoveUnreachableCoalesceArgumentsRule trims arguments based on type nullability metadata (e.g., COALESCE(nonNull, x) -> nonNull), without any join awareness.
2. This one is specific to joins and merging both would look less elegant IMO.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org