You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2020/11/27 13:42:19 UTC
[flink] branch release-1.12 updated:
[FLINK-20349][table-planner-blink] Fix checking for deadlock caused by
exchange
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.12 by this push:
new 2f655ec [FLINK-20349][table-planner-blink] Fix checking for deadlock caused by exchange
2f655ec is described below
commit 2f655ec254b6a2162fcfddaf76813c07a19d9ca1
Author: TsReaper <ts...@gmail.com>
AuthorDate: Fri Nov 27 21:34:22 2020 +0800
[FLINK-20349][table-planner-blink] Fix checking for deadlock caused by exchange
(cherry picked from commit 6709a5aa967be80620263a05ef5763c94d96aaf5)
---
.../processors/utils/InputOrderCalculator.java | 7 ++-
.../utils/InputPriorityConflictResolver.java | 34 +++++++++-
.../utils/InputPriorityGraphGenerator.java | 4 +-
.../utils/InputPriorityGraphGeneratorTest.java | 8 +--
.../planner/plan/batch/sql/DeadlockBreakupTest.xml | 38 ++++++++++++
.../plan/batch/sql/MultipleInputCreationTest.xml | 72 ++++++++++++++++++++++
.../plan/batch/sql/DeadlockBreakupTest.scala | 18 ++++++
.../plan/batch/sql/MultipleInputCreationTest.scala | 16 +++++
.../runtime/batch/sql/MultipleInputITCase.scala | 12 ++++
9 files changed, 201 insertions(+), 8 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputOrderCalculator.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputOrderCalculator.java
index ff2b685..086df98 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputOrderCalculator.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputOrderCalculator.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.plan.processors.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
@@ -167,6 +168,10 @@ public class InputOrderCalculator extends InputPriorityGraphGenerator {
@Override
protected void resolveInputPriorityConflict(ExecNode<?, ?> node, int higherInput, int lowerInput) {
- throw new IllegalStateException("A conflict is detected. This is unexpected.");
+ throw new IllegalStateException(
+ "A conflict is detected. This is a bug. Please file an issue.\n" +
+ "To work around this bug, please set " +
+ OptimizerConfigOptions.TABLE_OPTIMIZER_MULTIPLE_INPUT_ENABLED.key() +
+ " to false to disable multiple input operator.");
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityConflictResolver.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityConflictResolver.java
index 75a6672..4ef68fe 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityConflictResolver.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityConflictResolver.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.plan.processors.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecExchange;
@@ -67,7 +68,7 @@ public class InputPriorityConflictResolver extends InputPriorityGraphGenerator {
ExecNode<?, ?> lowerNode = node.getInputNodes().get(lowerInput);
if (lowerNode instanceof BatchExecExchange) {
BatchExecExchange exchange = (BatchExecExchange) lowerNode;
- if (higherNode == lowerNode) {
+ if (isConflictCausedByExchange(higherNode, exchange)) {
// special case: if exchange is exactly the reuse node,
// we should split it into two nodes
BatchExecExchange newExchange = exchange.copy(
@@ -84,6 +85,14 @@ public class InputPriorityConflictResolver extends InputPriorityGraphGenerator {
}
}
+ private boolean isConflictCausedByExchange(ExecNode<?, ?> higherNode, BatchExecExchange lowerNode) {
+ // check if `lowerNode` is the ancestor of `higherNode`,
+ // if yes then conflict is caused by `lowerNode`
+ ConflictCausedByExchangeChecker checker = new ConflictCausedByExchangeChecker(lowerNode);
+ checker.visit(higherNode);
+ return checker.found;
+ }
+
private BatchExecExchange createExchange(ExecNode<?, ?> node, int idx) {
RelNode inputRel = (RelNode) node.getInputNodes().get(idx);
@@ -109,4 +118,27 @@ public class InputPriorityConflictResolver extends InputPriorityGraphGenerator {
exchange.setRequiredShuffleMode(shuffleMode);
return exchange;
}
+
+ private static class ConflictCausedByExchangeChecker extends AbstractExecNodeExactlyOnceVisitor {
+
+ private final BatchExecExchange exchange;
+ private boolean found;
+
+ private ConflictCausedByExchangeChecker(BatchExecExchange exchange) {
+ this.exchange = exchange;
+ }
+
+ @Override
+ protected void visitNode(ExecNode<?, ?> node) {
+ if (node == exchange) {
+ found = true;
+ }
+ for (ExecNode<?, ?> inputNode : node.getInputNodes()) {
+ visit(inputNode);
+ if (found) {
+ return;
+ }
+ }
+ }
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGenerator.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGenerator.java
index 442055b..132ba73 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGenerator.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGenerator.java
@@ -156,7 +156,7 @@ public abstract class InputPriorityGraphGenerator {
private void addTopologyEdges(ExecNode<?, ?> node, int higherInput, int lowerInput) {
ExecNode<?, ?> higherNode = node.getInputNodes().get(higherInput);
ExecNode<?, ?> lowerNode = node.getInputNodes().get(lowerInput);
- List<ExecNode<?, ?>> lowerAncestors = calculateAncestors(lowerNode);
+ List<ExecNode<?, ?>> lowerAncestors = calculatePipelinedAncestors(lowerNode);
List<Tuple2<ExecNode<?, ?>, ExecNode<?, ?>>> linkedEdges = new ArrayList<>();
for (ExecNode<?, ?> ancestor : lowerAncestors) {
@@ -177,7 +177,7 @@ public abstract class InputPriorityGraphGenerator {
* Find the ancestors by going through PIPELINED edges.
*/
@VisibleForTesting
- List<ExecNode<?, ?>> calculateAncestors(ExecNode<?, ?> node) {
+ List<ExecNode<?, ?>> calculatePipelinedAncestors(ExecNode<?, ?> node) {
List<ExecNode<?, ?>> ret = new ArrayList<>();
AbstractExecNodeExactlyOnceVisitor ancestorVisitor = new AbstractExecNodeExactlyOnceVisitor() {
@Override
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGeneratorTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGeneratorTest.java
index 4b2c214..0090da1 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGeneratorTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/processors/utils/InputPriorityGraphGeneratorTest.java
@@ -36,7 +36,7 @@ import java.util.Set;
public class InputPriorityGraphGeneratorTest {
@Test
- public void testCalculateAncestors() {
+ public void testCalculatePipelinedAncestors() {
// P = ExecEdge.DamBehavior.PIPELINED, E = ExecEdge.DamBehavior.END_INPUT
//
// 0 ------P----> 1 -E--> 2
@@ -59,14 +59,14 @@ public class InputPriorityGraphGeneratorTest {
Collections.singletonList(nodes[2]),
Collections.emptySet(),
ExecEdge.DamBehavior.END_INPUT);
- List<ExecNode<?, ?>> ancestors = resolver.calculateAncestors(nodes[2]);
+ List<ExecNode<?, ?>> ancestors = resolver.calculatePipelinedAncestors(nodes[2]);
Assert.assertEquals(2, ancestors.size());
Assert.assertTrue(ancestors.contains(nodes[0]));
Assert.assertTrue(ancestors.contains(nodes[5]));
}
@Test
- public void testCalculateBoundedAncestors() {
+ public void testCalculateBoundedPipelinedAncestors() {
// P = ExecEdge.DamBehavior.PIPELINED, E = ExecEdge.DamBehavior.END_INPUT
//
// 0 -P-> 1 -P-> 2
@@ -84,7 +84,7 @@ public class InputPriorityGraphGeneratorTest {
Collections.singletonList(nodes[2]),
new HashSet<>(Collections.singleton(nodes[1])),
ExecEdge.DamBehavior.END_INPUT);
- List<ExecNode<?, ?>> ancestors = resolver.calculateAncestors(nodes[2]);
+ List<ExecNode<?, ?>> ancestors = resolver.calculatePipelinedAncestors(nodes[2]);
Assert.assertEquals(1, ancestors.size());
Assert.assertTrue(ancestors.contains(nodes[1]));
}
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.xml
index fafb127..79578a9 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.xml
@@ -460,6 +460,44 @@ HashJoin(joinType=[InnerJoin], where=[=(b, e0)], select=[a, b, d, e, a0, b0, d0,
]]>
</Resource>
</TestCase>
+ <TestCase name="testSubplanReuse_DeadlockCausedByReusingExchangeInAncestor">
+ <Resource name="sql">
+ <![CDATA[
+WITH T1 AS (
+ SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], a0=[$3], b0=[$4], c=[$5], k=[$6], v=[$7])
++- LogicalFilter(condition=[AND(>($0, 0), =($7, 0))])
+ +- LogicalJoin(condition=[=($0, $6)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], k=[$3], v=[+($1, $4)])
+ +- LogicalFilter(condition=[>($3, 0)])
+ +- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Calc(select=[a, b, a0, b0, c, k, CAST(0:BIGINT) AS v])
++- HashJoin(joinType=[InnerJoin], where=[=(a, k)], select=[a, b, a0, b0, c, k], build=[right])
+ :- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
+ : +- Calc(select=[a, b], where=[>(a, 0)], reuse_id=[2])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], reuse_id=[1])
+ +- Calc(select=[a, b, c, a0 AS k])
+ +- HashJoin(joinType=[InnerJoin], where=[AND(=(a, a0), =(+(b, b0), 0:BIGINT))], select=[a, b, c, a0, b0], build=[right])
+ :- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
+ : +- Reused(reference_id=[1])
+ +- Exchange(distribution=[hash[a]])
+ +- Reused(reference_id=[2])
+]]>
+ </Resource>
+ </TestCase>
<TestCase name="testSubplanReuse_SetExchangeAsBatch_OverAgg">
<Resource name="sql">
<![CDATA[
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.xml
index 07bbff3..cd58ff7 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.xml
@@ -405,6 +405,78 @@ Union(all=[true], union=[a, EXPR$1])
]]>
</Resource>
</TestCase>
+ <TestCase name="testDeadlockCausedByExchangeInAncestor[shuffleMode: ALL_EDGES_BLOCKING]">
+ <Resource name="sql">
+ <![CDATA[
+WITH T1 AS (
+ SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], a0=[$4], b0=[$5], c=[$6], nx=[$7], k=[$8], v=[$9])
++- LogicalFilter(condition=[AND(>($0, 0), =($9, 0))])
+ +- LogicalJoin(condition=[=($0, $8)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], nx=[$3], k=[$4], v=[+($1, $5)])
+ +- LogicalFilter(condition=[>($4, 0)])
+ +- LogicalJoin(condition=[=($0, $4)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Calc(select=[a, b, a0, b0, c, nx, k, CAST(0:BIGINT) AS v])
++- MultipleInput(readOrder=[2,1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[=(a, k)], select=[a, b, a0, b0, c, nx, k], build=[right])\n:- [#1] Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])\n+- Calc(select=[a, b, c, nx, a0 AS k])\n +- HashJoin(joinType=[InnerJoin], where=[AND(=(a, a0), =(+(b, b0), 0:BIGINT))], select=[a, b, c, nx, a0, b0], build=[right])\n :- [#2] Exchange(distribution=[hash[a]])\n +- [#3] Exchange(distribution=[hash[a]])\n])
+ :- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
+ : +- Calc(select=[a, b], where=[>(a, 0)], reuse_id=[2])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]], fields=[a, b, c, nx], reuse_id=[1])
+ :- Exchange(distribution=[hash[a]])
+ : +- Reused(reference_id=[1])
+ +- Exchange(distribution=[hash[a]])
+ +- Reused(reference_id=[2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testDeadlockCausedByExchangeInAncestor[shuffleMode: ALL_EDGES_PIPELINED]">
+ <Resource name="sql">
+ <![CDATA[
+WITH T1 AS (
+ SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], a0=[$4], b0=[$5], c=[$6], nx=[$7], k=[$8], v=[$9])
++- LogicalFilter(condition=[AND(>($0, 0), =($9, 0))])
+ +- LogicalJoin(condition=[=($0, $8)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], nx=[$3], k=[$4], v=[+($1, $5)])
+ +- LogicalFilter(condition=[>($4, 0)])
+ +- LogicalJoin(condition=[=($0, $4)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Calc(select=[a, b, a0, b0, c, nx, k, CAST(0:BIGINT) AS v])
++- MultipleInput(readOrder=[2,1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[=(a, k)], select=[a, b, a0, b0, c, nx, k], build=[right])\n:- [#1] Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])\n+- Calc(select=[a, b, c, nx, a0 AS k])\n +- HashJoin(joinType=[InnerJoin], where=[AND(=(a, a0), =(+(b, b0), 0:BIGINT))], select=[a, b, c, nx, a0, b0], build=[right])\n :- [#2] Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])\n +- [#3] Exchange(distribution=[hash[a]])\n])
+ :- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
+ : +- Calc(select=[a, b], where=[>(a, 0)], reuse_id=[2])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c, nx)]]], fields=[a, b, c, nx], reuse_id=[1])
+ :- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
+ : +- Reused(reference_id=[1])
+ +- Exchange(distribution=[hash[a]])
+ +- Reused(reference_id=[2])
+]]>
+ </Resource>
+ </TestCase>
<TestCase name="testNoPriorityConstraint[shuffleMode: ALL_EDGES_BLOCKING]">
<Resource name="sql">
<![CDATA[
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.scala
index c52b21d..abc193d 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DeadlockBreakupTest.scala
@@ -231,4 +231,22 @@ class DeadlockBreakupTest extends TableTestBase {
|""".stripMargin
util.verifyPlan(sqlQuery)
}
+
+ @Test
+ def testSubplanReuse_DeadlockCausedByReusingExchangeInAncestor(): Unit = {
+ util.tableEnv.getConfig.getConfiguration.setBoolean(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SUB_PLAN_ENABLED, true)
+ util.tableEnv.getConfig.getConfiguration.setBoolean(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_MULTIPLE_INPUT_ENABLED, false)
+ util.tableEnv.getConfig.getConfiguration.setString(
+ ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "NestedLoopJoin,SortMergeJoin")
+ val sqlQuery =
+ """
+ |WITH T1 AS (
+ | SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ | FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+ |SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+ |""".stripMargin
+ util.verifyPlan(sqlQuery)
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.scala
index ff29d6b..571bb7f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/MultipleInputCreationTest.scala
@@ -323,6 +323,22 @@ class MultipleInputCreationTest(shuffleMode: String) extends TableTestBase {
util.verifyPlan(sql)
}
+ @Test
+ def testDeadlockCausedByExchangeInAncestor(): Unit = {
+ util.tableEnv.getConfig.getConfiguration.setBoolean(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SOURCE_ENABLED, true)
+ util.tableEnv.getConfig.getConfiguration.setString(
+ ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "NestedLoopJoin,SortMergeJoin")
+ val sql =
+ """
+ |WITH T1 AS (
+ | SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ | FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+ |SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+ |""".stripMargin
+ util.verifyPlan(sql)
+ }
+
def createChainableTableSource(): Unit = {
val dataStream = util.getStreamEnv.fromSource(
new MockSource(Boundedness.BOUNDED, 1),
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/MultipleInputITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/MultipleInputITCase.scala
index ca4aa78..33707ce 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/MultipleInputITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/MultipleInputITCase.scala
@@ -170,6 +170,18 @@ class MultipleInputITCase(shuffleMode: String) extends BatchTestBase {
)
}
+ @Test
+ def testDeadlockCausedByExchangeInAncestor(): Unit = {
+ checkMultipleInputResult(
+ """
+ |WITH T1 AS (
+ | SELECT x1.*, x2.a AS k, (x1.b + x2.b) AS v
+ | FROM x x1 LEFT JOIN x x2 ON x1.a = x2.a WHERE x2.a > 0)
+ |SELECT x.a, x.b, T1.* FROM x LEFT JOIN T1 ON x.a = T1.k WHERE x.a > 0 AND T1.v = 0
+ |""".stripMargin
+ )
+ }
+
def checkMultipleInputResult(sql: String): Unit = {
tEnv.getConfig.getConfiguration.setBoolean(
OptimizerConfigOptions.TABLE_OPTIMIZER_MULTIPLE_INPUT_ENABLED, false)