You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "xuyangzhong (via GitHub)" <gi...@apache.org> on 2025/06/13 11:49:40 UTC
[PR] [FLINK-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
xuyangzhong opened a new pull request, #26679:
URL: https://github.com/apache/flink/pull/26679
## What is the purpose of the change
This pr tries to convert a regular join to a delta join for the simplest pattern in planner.
It requires:
- inner join
- only contains source, calc, join, sink, exchange
- no calc between source and join
- append-only source
## Brief change log
- *Introduce a new rule to do the optimization*
- *Add tests for it*
## Verifying this change
New tests are added to verity this pr.
## 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?
--
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2149558295
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeltaJoinTest.scala:
##########
@@ -0,0 +1,583 @@
+/*
+ * 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.batch.sql
+
+import org.apache.flink.table.api.{DataTypes, ExplainDetail, Schema}
+import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
+import org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
+import org.apache.flink.table.api.config.OptimizerConfigOptions.DeltaJoinStrategy
+import org.apache.flink.table.catalog.{CatalogTable, ObjectPath, ResolvedCatalogTable}
+import org.apache.flink.table.planner.JMap
+import org.apache.flink.table.planner.utils.{TableTestBase, TestingTableEnvironment}
+
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.{BeforeEach, Test}
+
+import java.util.{Collections, HashMap => JHashMap}
+
+/** Test for delta join. */
+class DeltaJoinTest extends TableTestBase {
+
+ private val util = streamTestUtil()
+ private val tEnv: TestingTableEnvironment = util.tableEnv.asInstanceOf[TestingTableEnvironment]
+
+ private val testComment = "test comment"
+ private val testValuesTableOptions: JMap[String, String] = {
+ val options = new JHashMap[String, String]()
+ options.put("connector", "values")
+ options.put("bounded", "false")
+ options.put("sink-insert-only", "false")
+ options.put("sink-changelog-mode-enforced", "I,UA,UB,D")
+ options.put("async", "true")
+ options
+ }
+
+ @BeforeEach
+ def setup(): Unit = {
+ util.tableConfig.set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.AUTO)
+
+ addTable(
+ "src1",
+ Schema
+ .newBuilder()
+ .column("a0", DataTypes.INT.notNull)
+ .column("a1", DataTypes.DOUBLE.notNull)
+ .column("a2", DataTypes.STRING)
+ .column("a3", DataTypes.INT)
+ .primaryKey("a0")
+ .index("a1", "a2")
+ .build()
+ )
+
+ addTable(
+ "src2",
+ Schema
+ .newBuilder()
+ .column("b0", DataTypes.INT.notNull)
+ .column("b2", DataTypes.STRING)
+ .column("b1", DataTypes.DOUBLE)
+ .primaryKey("b0")
+ .index("b2")
+ .build()
+ )
+
+ addTable(
+ "snk",
+ Schema
+ .newBuilder()
+ .column("l0", DataTypes.INT.notNull)
+ .column("l1", DataTypes.DOUBLE)
+ .column("l2", DataTypes.STRING)
+ .column("l3", DataTypes.INT)
+ .column("r0", DataTypes.INT.notNull)
+ .column("r2", DataTypes.STRING)
+ .column("r1", DataTypes.DOUBLE)
+ .primaryKey("l0", "r0")
+ .build()
+ )
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide2(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src1.a3 = src2.b0")
+ }
+
+ @Test
+ def testJoinKeysNotContainIndexOnOneSide(): Unit = {
+ util.verifyRelPlanInsert(
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2149149402
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeltaJoinTest.scala:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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.batch.sql
+
+import org.apache.flink.table.api.{DataTypes, ExplainDetail, Schema}
+import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
+import org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
+import org.apache.flink.table.api.config.OptimizerConfigOptions.DeltaJoinStrategy
+import org.apache.flink.table.catalog.{CatalogTable, ObjectPath, ResolvedCatalogTable}
+import org.apache.flink.table.planner.JMap
+import org.apache.flink.table.planner.utils.{TableTestBase, TestingTableEnvironment}
+
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.{BeforeEach, Test}
+
+import java.util.{Collections, HashMap => JHashMap}
+
+/** Test for delta join. */
+class DeltaJoinTest extends TableTestBase {
+
+ private val util = streamTestUtil()
+ private val tEnv: TestingTableEnvironment = util.tableEnv.asInstanceOf[TestingTableEnvironment]
+
+ private val testComment = "test comment"
+ private val testValuesTableOptions: JMap[String, String] = {
+ val options = new JHashMap[String, String]()
+ options.put("connector", "values")
+ options.put("bounded", "false")
+ options.put("sink-insert-only", "false")
+ options.put("sink-changelog-mode-enforced", "I,UA,UB,D")
+ options.put("async", "true")
+ options
+ }
+
+ @BeforeEach
+ def setup(): Unit = {
+ util.tableConfig.set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.AUTO)
+
+ addTable(
+ "src1",
+ Schema
+ .newBuilder()
+ .column("a0", DataTypes.INT.notNull)
+ .column("a1", DataTypes.DOUBLE.notNull)
+ .column("a2", DataTypes.STRING)
+ .column("a3", DataTypes.INT)
+ .primaryKey("a0")
+ .index("a1", "a2")
+ .build()
+ )
+
+ addTable(
+ "src2",
+ Schema
+ .newBuilder()
+ .column("b0", DataTypes.INT.notNull)
+ .column("b2", DataTypes.STRING)
+ .column("b1", DataTypes.DOUBLE)
+ .primaryKey("b0")
+ .index("b2")
+ .build()
+ )
+
+ addTable(
+ "snk",
+ Schema
+ .newBuilder()
+ .column("l0", DataTypes.INT.notNull)
+ .column("l1", DataTypes.DOUBLE)
+ .column("l2", DataTypes.STRING)
+ .column("l3", DataTypes.INT)
+ .column("r0", DataTypes.INT.notNull)
+ .column("r2", DataTypes.STRING)
+ .column("r1", DataTypes.DOUBLE)
+ .primaryKey("l0", "r0")
+ .build()
+ )
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide2(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src1.a3 = src2.b0")
+ }
+
+ @Test
+ def testJoinKeysNotContainIndexOnOneSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testWithNonEquiCondition1(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0")
+ }
+
+ @Test
+ def testWithNonEquiCondition2(): Unit = {
+ // could not optimize into delta join because there is a calc between join and source
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0 " +
+ "and src2.b2 <> 'Hello' " +
+ "and src1.a0 > 99")
+ }
+
+ @Test
+ def testJsonPlanWithTableHints(): Unit = {
+ assertThatThrownBy(
+ () => {
+ util.verifyJsonPlan(
+ "insert into snk select * from src1 /*+ OPTIONS('failing-source'='true') */" +
+ "join src2 /*+ OPTIONS('failing-source'='true') */" +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0")
+ }).hasMessage("Introduce delta join in runtime later")
+ }
+
+ @Test
+ def testFilterProjectFieldsAfterJoin(): Unit = {
+ // could not optimize into delta join because there is a calc between sink and join
Review Comment:
Refer to the description of FLINK-37897, maybe we have supported the condition when there is a calc between sink and join.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/DeltaJoinRewriteRule.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DeltaJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin;
+import org.apache.flink.table.planner.plan.utils.DeltaJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinTypeUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.immutables.value.Value;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/**
+ * A rule that converts a {@link StreamPhysicalJoin} to a {@link StreamPhysicalDeltaJoin} if all
+ * conditions are matched.
+ *
+ * <p>Currently, only {@link StreamPhysicalJoin} satisfied the following requirements can be
+ * converted to {@link StreamPhysicalDeltaJoin}.
+ *
+ * <ol>
+ * <li>The join is INNER join.
+ * <li>There is at least one join key pair in the join.
+ * <li>The downstream nodes of this join can accept duplicate changes.
+ * <li>All join inputs are insert only streams.
+ * <li>All downstream nodes of this join are in {@code
+ * DeltaJoinUtil#ALL_SUPPORTED_DELTA_JOIN_UPSTREAM_NODES}
+ * <li>The join keys include at least one complete index in each the source table in join inputs.
+ * <li>All table scans in this join inputs support async {@link LookupTableSource}.
+ * </ol>
+ *
+ * <p>See more at {@link DeltaJoinUtil#canConvertToDeltaJoin}.
+ */
+@Value.Enclosing
+public class DeltaJoinRewriteRule extends RelRule<DeltaJoinRewriteRule.Config> {
+
+ public static final DeltaJoinRewriteRule INSTANCE =
+ DeltaJoinRewriteRule.Config.DEFAULT.toRule();
+
+ private DeltaJoinRewriteRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ TableConfig tableConfig = unwrapTableConfig(call);
+ OptimizerConfigOptions.DeltaJoinStrategy deltaJoinStrategy =
+ tableConfig.get(OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY);
+ if (OptimizerConfigOptions.DeltaJoinStrategy.NONE == deltaJoinStrategy) {
+ return false;
+ }
+
+ StreamPhysicalJoin join = call.rel(0);
+ return DeltaJoinUtil.canConvertToDeltaJoin(join);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ StreamPhysicalJoin join = call.rel(0);
+ StreamPhysicalDeltaJoin deltaJoin = convertToDeltaJoin(join);
+ call.transformTo(deltaJoin);
+ }
+
+ private StreamPhysicalDeltaJoin convertToDeltaJoin(StreamPhysicalJoin join) {
+ DeltaJoinSpec lookupRightJoinSpec = DeltaJoinUtil.getDeltaJoinSpec(join, true);
Review Comment:
If we set `treatRightAsLookupSide` is true, maybe the `DeltaJoinSpec` we get should be named `lookupLeftJoinSpec`.
--
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong merged PR #26679:
URL: https://github.com/apache/flink/pull/26679
--
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2149318852
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeltaJoinTest.scala:
##########
@@ -0,0 +1,583 @@
+/*
+ * 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.batch.sql
+
+import org.apache.flink.table.api.{DataTypes, ExplainDetail, Schema}
+import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
+import org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
+import org.apache.flink.table.api.config.OptimizerConfigOptions.DeltaJoinStrategy
+import org.apache.flink.table.catalog.{CatalogTable, ObjectPath, ResolvedCatalogTable}
+import org.apache.flink.table.planner.JMap
+import org.apache.flink.table.planner.utils.{TableTestBase, TestingTableEnvironment}
+
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.{BeforeEach, Test}
+
+import java.util.{Collections, HashMap => JHashMap}
+
+/** Test for delta join. */
+class DeltaJoinTest extends TableTestBase {
+
+ private val util = streamTestUtil()
+ private val tEnv: TestingTableEnvironment = util.tableEnv.asInstanceOf[TestingTableEnvironment]
+
+ private val testComment = "test comment"
+ private val testValuesTableOptions: JMap[String, String] = {
+ val options = new JHashMap[String, String]()
+ options.put("connector", "values")
+ options.put("bounded", "false")
+ options.put("sink-insert-only", "false")
+ options.put("sink-changelog-mode-enforced", "I,UA,UB,D")
+ options.put("async", "true")
+ options
+ }
+
+ @BeforeEach
+ def setup(): Unit = {
+ util.tableConfig.set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.AUTO)
+
+ addTable(
+ "src1",
+ Schema
+ .newBuilder()
+ .column("a0", DataTypes.INT.notNull)
+ .column("a1", DataTypes.DOUBLE.notNull)
+ .column("a2", DataTypes.STRING)
+ .column("a3", DataTypes.INT)
+ .primaryKey("a0")
+ .index("a1", "a2")
+ .build()
+ )
+
+ addTable(
+ "src2",
+ Schema
+ .newBuilder()
+ .column("b0", DataTypes.INT.notNull)
+ .column("b2", DataTypes.STRING)
+ .column("b1", DataTypes.DOUBLE)
+ .primaryKey("b0")
+ .index("b2")
+ .build()
+ )
+
+ addTable(
+ "snk",
+ Schema
+ .newBuilder()
+ .column("l0", DataTypes.INT.notNull)
+ .column("l1", DataTypes.DOUBLE)
+ .column("l2", DataTypes.STRING)
+ .column("l3", DataTypes.INT)
+ .column("r0", DataTypes.INT.notNull)
+ .column("r2", DataTypes.STRING)
+ .column("r1", DataTypes.DOUBLE)
+ .primaryKey("l0", "r0")
+ .build()
+ )
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide2(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src1.a3 = src2.b0")
+ }
+
+ @Test
+ def testJoinKeysNotContainIndexOnOneSide(): Unit = {
+ util.verifyRelPlanInsert(
Review Comment:
We can add notes`// could not optimize into delta join because join keys do not contain indexes strictly` 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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26679:
URL: https://github.com/apache/flink/pull/26679#issuecomment-2970156280
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f1f156c1f52480abc835faa14669f6c9fbc0b918",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f1f156c1f52480abc835faa14669f6c9fbc0b918",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f1f156c1f52480abc835faa14669f6c9fbc0b918 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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "lincoln-lil (via GitHub)" <gi...@apache.org>.
lincoln-lil commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2151286553
##########
docs/layouts/shortcodes/generated/optimizer_config_configuration.html:
##########
@@ -29,6 +29,12 @@
<td>Integer</td>
<td>The maximum number of joined nodes allowed in the bushy join reorder algorithm, otherwise the left-deep join reorder algorithm will be used. The search space of bushy join reorder algorithm will increase with the increase of this threshold value, so this threshold is not recommended to be set too large. The default value is 12.</td>
</tr>
+ <tr>
+ <td><h5>table.optimizer.delta-join.strategy</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">AUTO</td>
+ <td><p>Enum</p></td>
+ <td>Whether to enable delta join strategy. Only AUTO, FORCE or NONE can be set. Default it AUTO.<br /><br />Possible values:<ul><li>"AUTO": Optimizer will try to use delta join first. If it fails, it will fallback to the regular join.</li><li>"FORCE": Use the delta join. If it fails, an exception will be thrown.</li><li>"NONE": Don't try to use delta join.</li></ul></td>
Review Comment:
Since it's a strategy, we can rephrase to describe the strategies, e.g., "Strategy for optimizing the delta-join", WDYT?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/DeltaJoinRewriteRule.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DeltaJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin;
+import org.apache.flink.table.planner.plan.utils.DeltaJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinTypeUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.immutables.value.Value;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/**
+ * A rule that converts a {@link StreamPhysicalJoin} to a {@link StreamPhysicalDeltaJoin} if all
+ * conditions are matched.
+ *
+ * <p>Currently, only {@link StreamPhysicalJoin} satisfied the following requirements can be
+ * converted to {@link StreamPhysicalDeltaJoin}.
+ *
+ * <ol>
+ * <li>The join is INNER join.
+ * <li>There is at least one join key pair in the join.
+ * <li>The downstream nodes of this join can accept duplicate changes.
+ * <li>All join inputs are insert only streams.
+ * <li>All downstream nodes of this join are in {@code
+ * DeltaJoinUtil#ALL_SUPPORTED_DELTA_JOIN_UPSTREAM_NODES}
+ * <li>The join keys include at least one complete index in each the source table in join inputs.
Review Comment:
nit: "... in each source table of the join input"
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/DeltaJoinRewriteRule.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DeltaJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin;
+import org.apache.flink.table.planner.plan.utils.DeltaJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinTypeUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.immutables.value.Value;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/**
+ * A rule that converts a {@link StreamPhysicalJoin} to a {@link StreamPhysicalDeltaJoin} if all
+ * conditions are matched.
+ *
+ * <p>Currently, only {@link StreamPhysicalJoin} satisfied the following requirements can be
+ * converted to {@link StreamPhysicalDeltaJoin}.
+ *
+ * <ol>
+ * <li>The join is INNER join.
+ * <li>There is at least one join key pair in the join.
+ * <li>The downstream nodes of this join can accept duplicate changes.
+ * <li>All join inputs are insert only streams.
+ * <li>All downstream nodes of this join are in {@code
+ * DeltaJoinUtil#ALL_SUPPORTED_DELTA_JOIN_UPSTREAM_NODES}
+ * <li>The join keys include at least one complete index in each the source table in join inputs.
+ * <li>All table scans in this join inputs support async {@link LookupTableSource}.
Review Comment:
-> "All table sources of this join inputs" ?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/DeltaJoinSpec.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.nodes.exec.spec;
+
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.calcite.rex.RexNode;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * DeltaJoinSpec describes one side lookup the dim table on the other side.
Review Comment:
How about 'DeltaJoinSpec defines how one side looks up the dimension table on the other side' ?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala:
##########
@@ -520,7 +520,9 @@ object FlinkStreamRuleSets {
// incremental agg rule
IncrementalAggregateRule.INSTANCE,
// optimize window agg rule
- TwoStageOptimizedWindowAggregateRule.INSTANCE
+ TwoStageOptimizedWindowAggregateRule.INSTANCE,
+ // delta join redundant data acceptable infer rule
Review Comment:
Just "delta join rule" is enough
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/DeltaJoinSpec.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.nodes.exec.spec;
+
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.calcite.rex.RexNode;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * DeltaJoinSpec describes one side lookup the dim table on the other side.
+ *
+ * <p>This class corresponds to {@link StreamPhysicalDeltaJoin}.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DeltaJoinSpec {
+
+ public static final String FIELD_NAME_LOOKUP_TABLE = "lookupTable";
+ public static final String FIELD_NAME_LOOKUP_KEYS = "lookupKeys";
+ public static final String FIELD_NAME_REMAINING_CONDITION = "remainingCondition";
+
+ @JsonProperty(FIELD_NAME_LOOKUP_TABLE)
+ private final TemporalTableSourceSpec lookupTable;
+
+ // <lookup column index of the dim table, stream side's related key>
Review Comment:
nit: Use a consistent commenting style
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/DeltaJoinRewriteRule.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DeltaJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin;
+import org.apache.flink.table.planner.plan.utils.DeltaJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinTypeUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.immutables.value.Value;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/**
+ * A rule that converts a {@link StreamPhysicalJoin} to a {@link StreamPhysicalDeltaJoin} if all
+ * conditions are matched.
+ *
+ * <p>Currently, only {@link StreamPhysicalJoin} satisfied the following requirements can be
+ * converted to {@link StreamPhysicalDeltaJoin}.
+ *
+ * <ol>
+ * <li>The join is INNER join.
+ * <li>There is at least one join key pair in the join.
+ * <li>The downstream nodes of this join can accept duplicate changes.
+ * <li>All join inputs are insert only streams.
+ * <li>All downstream nodes of this join are in {@code
Review Comment:
Should be 'upstream' ?
--
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2149284824
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeltaJoinTest.scala:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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.batch.sql
+
+import org.apache.flink.table.api.{DataTypes, ExplainDetail, Schema}
+import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
+import org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
+import org.apache.flink.table.api.config.OptimizerConfigOptions.DeltaJoinStrategy
+import org.apache.flink.table.catalog.{CatalogTable, ObjectPath, ResolvedCatalogTable}
+import org.apache.flink.table.planner.JMap
+import org.apache.flink.table.planner.utils.{TableTestBase, TestingTableEnvironment}
+
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.{BeforeEach, Test}
+
+import java.util.{Collections, HashMap => JHashMap}
+
+/** Test for delta join. */
+class DeltaJoinTest extends TableTestBase {
+
+ private val util = streamTestUtil()
+ private val tEnv: TestingTableEnvironment = util.tableEnv.asInstanceOf[TestingTableEnvironment]
+
+ private val testComment = "test comment"
+ private val testValuesTableOptions: JMap[String, String] = {
+ val options = new JHashMap[String, String]()
+ options.put("connector", "values")
+ options.put("bounded", "false")
+ options.put("sink-insert-only", "false")
+ options.put("sink-changelog-mode-enforced", "I,UA,UB,D")
+ options.put("async", "true")
+ options
+ }
+
+ @BeforeEach
+ def setup(): Unit = {
+ util.tableConfig.set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.AUTO)
+
+ addTable(
+ "src1",
+ Schema
+ .newBuilder()
+ .column("a0", DataTypes.INT.notNull)
+ .column("a1", DataTypes.DOUBLE.notNull)
+ .column("a2", DataTypes.STRING)
+ .column("a3", DataTypes.INT)
+ .primaryKey("a0")
+ .index("a1", "a2")
+ .build()
+ )
+
+ addTable(
+ "src2",
+ Schema
+ .newBuilder()
+ .column("b0", DataTypes.INT.notNull)
+ .column("b2", DataTypes.STRING)
+ .column("b1", DataTypes.DOUBLE)
+ .primaryKey("b0")
+ .index("b2")
+ .build()
+ )
+
+ addTable(
+ "snk",
+ Schema
+ .newBuilder()
+ .column("l0", DataTypes.INT.notNull)
+ .column("l1", DataTypes.DOUBLE)
+ .column("l2", DataTypes.STRING)
+ .column("l3", DataTypes.INT)
+ .column("r0", DataTypes.INT.notNull)
+ .column("r2", DataTypes.STRING)
+ .column("r1", DataTypes.DOUBLE)
+ .primaryKey("l0", "r0")
+ .build()
+ )
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testJoinKeysContainIndexOnBothSide2(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src1.a3 = src2.b0")
+ }
+
+ @Test
+ def testJoinKeysNotContainIndexOnOneSide(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a2 = src2.b2")
+ }
+
+ @Test
+ def testWithNonEquiCondition1(): Unit = {
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0")
+ }
+
+ @Test
+ def testWithNonEquiCondition2(): Unit = {
+ // could not optimize into delta join because there is a calc between join and source
+ util.verifyRelPlanInsert(
+ "insert into snk select * from src1 join src2 " +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0 " +
+ "and src2.b2 <> 'Hello' " +
+ "and src1.a0 > 99")
+ }
+
+ @Test
+ def testJsonPlanWithTableHints(): Unit = {
+ assertThatThrownBy(
+ () => {
+ util.verifyJsonPlan(
+ "insert into snk select * from src1 /*+ OPTIONS('failing-source'='true') */" +
+ "join src2 /*+ OPTIONS('failing-source'='true') */" +
+ "on src1.a1 = src2.b1 " +
+ "and src1.a2 = src2.b2 " +
+ "and src2.b0 > src1.a0")
+ }).hasMessage("Introduce delta join in runtime later")
+ }
+
+ @Test
+ def testFilterProjectFieldsAfterJoin(): Unit = {
+ // could not optimize into delta join because there is a calc between sink and join
Review Comment:
You're right! I'll update the tests to split it into different situations.
--
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-37897][table-planner] Support converting a regular join to a delta join for the simplest pattern in planner [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26679:
URL: https://github.com/apache/flink/pull/26679#discussion_r2149293042
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/DeltaJoinRewriteRule.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.physical.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DeltaJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin;
+import org.apache.flink.table.planner.plan.utils.DeltaJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinTypeUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.immutables.value.Value;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/**
+ * A rule that converts a {@link StreamPhysicalJoin} to a {@link StreamPhysicalDeltaJoin} if all
+ * conditions are matched.
+ *
+ * <p>Currently, only {@link StreamPhysicalJoin} satisfied the following requirements can be
+ * converted to {@link StreamPhysicalDeltaJoin}.
+ *
+ * <ol>
+ * <li>The join is INNER join.
+ * <li>There is at least one join key pair in the join.
+ * <li>The downstream nodes of this join can accept duplicate changes.
+ * <li>All join inputs are insert only streams.
+ * <li>All downstream nodes of this join are in {@code
+ * DeltaJoinUtil#ALL_SUPPORTED_DELTA_JOIN_UPSTREAM_NODES}
+ * <li>The join keys include at least one complete index in each the source table in join inputs.
+ * <li>All table scans in this join inputs support async {@link LookupTableSource}.
+ * </ol>
+ *
+ * <p>See more at {@link DeltaJoinUtil#canConvertToDeltaJoin}.
+ */
+@Value.Enclosing
+public class DeltaJoinRewriteRule extends RelRule<DeltaJoinRewriteRule.Config> {
+
+ public static final DeltaJoinRewriteRule INSTANCE =
+ DeltaJoinRewriteRule.Config.DEFAULT.toRule();
+
+ private DeltaJoinRewriteRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ TableConfig tableConfig = unwrapTableConfig(call);
+ OptimizerConfigOptions.DeltaJoinStrategy deltaJoinStrategy =
+ tableConfig.get(OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY);
+ if (OptimizerConfigOptions.DeltaJoinStrategy.NONE == deltaJoinStrategy) {
+ return false;
+ }
+
+ StreamPhysicalJoin join = call.rel(0);
+ return DeltaJoinUtil.canConvertToDeltaJoin(join);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ StreamPhysicalJoin join = call.rel(0);
+ StreamPhysicalDeltaJoin deltaJoin = convertToDeltaJoin(join);
+ call.transformTo(deltaJoin);
+ }
+
+ private StreamPhysicalDeltaJoin convertToDeltaJoin(StreamPhysicalJoin join) {
+ DeltaJoinSpec lookupRightJoinSpec = DeltaJoinUtil.getDeltaJoinSpec(join, true);
Review Comment:
Fixed.
--
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