You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/08/30 13:08:04 UTC
[flink] 02/02: [FlINK-32865][table-planner] Make ExecutionOrderEnforcer support OFCG
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 94ba97a13abc7e70a55d929fd5fdf85d52417639
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Thu Aug 24 16:02:32 2023 +0800
[FlINK-32865][table-planner] Make ExecutionOrderEnforcer support OFCG
This closes #23282
---
.../batch/BatchExecExecutionOrderEnforcer.java | 29 ++++++++
.../ExecutionOrderEnforcerFusionCodegenSpec.scala | 78 ++++++++++++++++++++++
.../runtime/batch/sql/DynamicFilteringITCase.java | 53 ++++++++++-----
3 files changed, 143 insertions(+), 17 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExecutionOrderEnforcer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExecutionOrderEnforcer.java
index c0ba1fd9335..e52fcc43989 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExecutionOrderEnforcer.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExecutionOrderEnforcer.java
@@ -26,6 +26,9 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
import org.apache.flink.table.planner.codegen.dynamicfiltering.ExecutionOrderEnforcerCodeGenerator;
import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.fusion.OpFusionCodegenSpecGenerator;
+import org.apache.flink.table.planner.plan.fusion.generator.TwoInputOpFusionCodegenSpecGenerator;
+import org.apache.flink.table.planner.plan.fusion.spec.ExecutionOrderEnforcerFusionCodegenSpec;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
@@ -70,6 +73,32 @@ public class BatchExecExecutionOrderEnforcer extends ExecNodeBase<RowData>
description);
}
+ @Override
+ public boolean supportFusionCodegen() {
+ return true;
+ }
+
+ @Override
+ protected OpFusionCodegenSpecGenerator translateToFusionCodegenSpecInternal(
+ PlannerBase planner, ExecNodeConfig config) {
+ OpFusionCodegenSpecGenerator leftInput =
+ getInputEdges().get(0).translateToFusionCodegenSpec(planner);
+ OpFusionCodegenSpecGenerator rightInput =
+ getInputEdges().get(1).translateToFusionCodegenSpec(planner);
+ OpFusionCodegenSpecGenerator runtimeFilterGenerator =
+ new TwoInputOpFusionCodegenSpecGenerator(
+ leftInput,
+ rightInput,
+ 0L,
+ (RowType) getOutputType(),
+ new ExecutionOrderEnforcerFusionCodegenSpec(
+ new CodeGeneratorContext(
+ config, planner.getFlinkContext().getClassLoader())));
+ leftInput.addOutput(1, runtimeFilterGenerator);
+ rightInput.addOutput(2, runtimeFilterGenerator);
+ return runtimeFilterGenerator;
+ }
+
@Override
@SuppressWarnings("unchecked")
protected Transformation<RowData> translateToPlanInternal(
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/fusion/spec/ExecutionOrderEnforcerFusionCodegenSpec.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/fusion/spec/ExecutionOrderEnforcerFusionCodegenSpec.scala
new file mode 100644
index 00000000000..ae1d434d5d7
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/fusion/spec/ExecutionOrderEnforcerFusionCodegenSpec.scala
@@ -0,0 +1,78 @@
+/*
+ * 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.fusion.spec
+
+import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, GeneratedExpression}
+import org.apache.flink.table.planner.plan.fusion.{OpFusionCodegenSpecBase, OpFusionContext}
+
+import java.util
+
+/** The operator fusion codegen spec for ExecutionOrderEnforcer. */
+class ExecutionOrderEnforcerFusionCodegenSpec(opCodegenCtx: CodeGeneratorContext)
+ extends OpFusionCodegenSpecBase(opCodegenCtx) {
+ private lazy val sourceInputId = 2
+
+ private var dynamicFilteringInputContext: OpFusionContext = _
+ private var sourceInputContext: OpFusionContext = _
+
+ override def setup(opFusionContext: OpFusionContext): Unit = {
+ super.setup(opFusionContext)
+ val inputContexts = fusionContext.getInputFusionContexts
+ assert(inputContexts.size == 2)
+ dynamicFilteringInputContext = inputContexts.get(0)
+ sourceInputContext = inputContexts.get(1)
+ }
+
+ override def variablePrefix(): String = "orderEnforcer"
+
+ override def doProcessProduce(codegenCtx: CodeGeneratorContext): Unit = {
+ dynamicFilteringInputContext.processProduce(codegenCtx)
+ sourceInputContext.processProduce(codegenCtx)
+ }
+
+ override def doEndInputProduce(codegenCtx: CodeGeneratorContext): Unit = {
+ dynamicFilteringInputContext.endInputProduce(codegenCtx)
+ sourceInputContext.endInputProduce(codegenCtx)
+ }
+
+ override def doProcessConsume(
+ inputId: Int,
+ inputVars: util.List[GeneratedExpression],
+ row: GeneratedExpression): String = {
+ if (inputId == sourceInputId) {
+ s"""
+ |// call downstream to consume the row
+ |${row.code}
+ |${fusionContext.processConsume(null, row.resultTerm)}
+ |""".stripMargin
+ } else {
+ ""
+ }
+ }
+
+ override def doEndInputConsume(inputId: Int): String = {
+ if (inputId == sourceInputId) {
+ s"""
+ |// call downstream endInput method
+ |${fusionContext.endInputConsume()}
+ |""".stripMargin
+ } else {
+ ""
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/DynamicFilteringITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/DynamicFilteringITCase.java
index c5ca3a83107..fe61487e4d1 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/DynamicFilteringITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/DynamicFilteringITCase.java
@@ -48,8 +48,10 @@ public class DynamicFilteringITCase extends BatchTestBase {
static Stream<Arguments> parameters() {
return Stream.of(
- Arguments.of(BatchShuffleMode.ALL_EXCHANGES_BLOCKING),
- Arguments.of(BatchShuffleMode.ALL_EXCHANGES_PIPELINED));
+ Arguments.of(BatchShuffleMode.ALL_EXCHANGES_BLOCKING, true),
+ Arguments.of(BatchShuffleMode.ALL_EXCHANGES_BLOCKING, false),
+ Arguments.of(BatchShuffleMode.ALL_EXCHANGES_PIPELINED, true),
+ Arguments.of(BatchShuffleMode.ALL_EXCHANGES_PIPELINED, false));
}
@BeforeEach
@@ -124,10 +126,12 @@ public class DynamicFilteringITCase extends BatchTestBase {
dataId3));
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testSimpleDynamicFiltering(BatchShuffleMode shuffleMode) {
+ public void testSimpleDynamicFiltering(BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact1, dim WHERE x = a AND z = 2",
JavaScalaConversionUtil.toScala(
@@ -153,11 +157,13 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testDynamicFilteringChainWithMultipleInput(BatchShuffleMode shuffleMode)
- throws Exception {
+ public void testDynamicFilteringChainWithMultipleInput(
+ BatchShuffleMode shuffleMode, boolean ofcg) throws Exception {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
String dataId1 = TestValuesTableFactory.registerData(TestData.data7());
tEnv.executeSql(
String.format(
@@ -200,10 +206,13 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testDynamicFilteringCannotChainWithMultipleInput(BatchShuffleMode shuffleMode) {
+ public void testDynamicFilteringCannotChainWithMultipleInput(
+ BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact1, dim, fact2 WHERE x = fact1.a and fact2.a = fact1.a AND z = 1 and fact1.e = 2 and fact2.e = 1",
JavaScalaConversionUtil.toScala(
@@ -231,10 +240,12 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testReuseDimSide(BatchShuffleMode shuffleMode) {
+ public void testReuseDimSide(BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact1, dim WHERE x = a AND z = 1 and b = 3"
+ "UNION ALL "
@@ -246,10 +257,13 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testDynamicFilteringWithStaticPartitionPruning(BatchShuffleMode shuffleMode) {
+ public void testDynamicFilteringWithStaticPartitionPruning(
+ BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact2, dim WHERE x = a and e = z AND y < 5 and a = 3",
JavaScalaConversionUtil.toScala(
@@ -260,10 +274,12 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testMultiplePartitionKeysWithFullKey(BatchShuffleMode shuffleMode) {
+ public void testMultiplePartitionKeysWithFullKey(BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact2, dim WHERE x = a AND z = e and y = 1",
JavaScalaConversionUtil.toScala(
@@ -271,10 +287,13 @@ public class DynamicFilteringITCase extends BatchTestBase {
false);
}
- @ParameterizedTest(name = "mode = {0}")
+ @ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
- public void testMultiplePartitionKeysWithPartialKey(BatchShuffleMode shuffleMode) {
+ public void testMultiplePartitionKeysWithPartialKey(
+ BatchShuffleMode shuffleMode, boolean ofcg) {
configBatchShuffleMode(tEnv.getConfig(), shuffleMode);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLED, ofcg);
checkResult(
"SELECT * FROM fact2, dim WHERE z = e and y = 1",
JavaScalaConversionUtil.toScala(