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 2021/01/15 05:44:26 UTC
[flink] 02/02: [FLINK-20876][table-planner-blink] Apply JoinSpec to
other join nodes and clean up parameters.
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 38327866376c1b55ffba9a8b4fdc390a0bfe0e10
Author: 龙三 <we...@alibaba-inc.com>
AuthorDate: Fri Jan 15 10:48:03 2021 +0800
[FLINK-20876][table-planner-blink] Apply JoinSpec to other join nodes and clean up parameters.
This closes #14606
---
.../plan/nodes/exec/batch/BatchExecHashJoin.java | 48 ++++++-----------
.../nodes/exec/batch/BatchExecNestedLoopJoin.java | 7 +--
.../nodes/exec/batch/BatchExecSortMergeJoin.java | 7 +--
.../plan/nodes/common/CommonPhysicalJoin.scala | 18 +------
.../physical/batch/BatchPhysicalHashJoin.scala | 42 +++++++--------
.../batch/BatchPhysicalNestedLoopJoin.scala | 10 ++--
.../batch/BatchPhysicalSortMergeJoin.scala | 22 +++++---
.../nodes/physical/stream/StreamExecJoin.scala | 31 +++++------
.../FlinkChangelogModeInferenceProgram.scala | 4 +-
.../flink/table/planner/plan/utils/JoinUtil.scala | 63 +---------------------
10 files changed, 81 insertions(+), 171 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashJoin.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashJoin.java
index 726b25f..b749e14 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashJoin.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashJoin.java
@@ -33,6 +33,7 @@ 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.exec.ExecNodeBase;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
import org.apache.flink.table.planner.plan.utils.JoinUtil;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.generated.GeneratedProjection;
@@ -43,24 +44,13 @@ import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.calcite.rex.RexNode;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
+import java.util.Arrays;
import java.util.stream.IntStream;
-import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/** {@link BatchExecNode} for Hash Join. */
public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExecNode<RowData> {
- private final FlinkJoinType joinType;
- private final int[] leftKeys;
- private final int[] rightKeys;
- private final boolean[] filterNulls;
- private final @Nullable RexNode nonEquiCondition;
+ private final JoinSpec joinSpec;
private final boolean leftIsBuild;
private final int estimatedLeftAvgRowSize;
private final int estimatedRightAvgRowSize;
@@ -69,29 +59,19 @@ public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExe
private final boolean tryDistinctBuildRow;
public BatchExecHashJoin(
- FlinkJoinType joinType,
- int[] leftKeys,
- int[] rightKeys,
- boolean[] filterNulls,
- @Nullable RexNode nonEquiCondition,
+ JoinSpec joinSpec,
int estimatedLeftAvgRowSize,
int estimatedRightAvgRowSize,
long estimatedLeftRowCount,
long estimatedRightRowCount,
boolean leftIsBuild,
boolean tryDistinctBuildRow,
- List<ExecEdge> inputEdges,
+ ExecEdge leftEdge,
+ ExecEdge rightEdge,
RowType outputType,
String description) {
- super(inputEdges, outputType, description);
- this.joinType = checkNotNull(joinType);
- this.leftKeys = checkNotNull(leftKeys);
- this.rightKeys = checkNotNull(rightKeys);
- this.filterNulls = checkNotNull(filterNulls);
- checkArgument(leftKeys.length > 0 && leftKeys.length == rightKeys.length);
- checkArgument(leftKeys.length == filterNulls.length);
-
- this.nonEquiCondition = nonEquiCondition;
+ super(Arrays.asList(leftEdge, rightEdge), outputType, description);
+ this.joinSpec = joinSpec;
this.leftIsBuild = leftIsBuild;
this.estimatedLeftAvgRowSize = estimatedLeftAvgRowSize;
this.estimatedRightAvgRowSize = estimatedRightAvgRowSize;
@@ -112,13 +92,18 @@ public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExe
RowType leftType = (RowType) leftInputNode.getOutputType();
RowType rightType = (RowType) rightInputNode.getOutputType();
+ JoinUtil.validateJoinSpec(joinSpec, leftType, rightType, false);
+
+ int[] leftKeys = joinSpec.getLeftKeys();
+ int[] rightKeys = joinSpec.getRightKeys();
LogicalType[] keyFieldTypes =
IntStream.of(leftKeys).mapToObj(leftType::getTypeAt).toArray(LogicalType[]::new);
RowType keyType = RowType.of(keyFieldTypes);
TableConfig config = planner.getTableConfig();
GeneratedJoinCondition condFunc =
- JoinUtil.generateConditionFunction(config, nonEquiCondition, leftType, rightType);
+ JoinUtil.generateConditionFunction(
+ config, joinSpec.getNonEquiCondition().orElse(null), leftType, rightType);
// projection for equals
GeneratedProjection leftProj =
@@ -178,6 +163,7 @@ public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExe
// operator
StreamOperatorFactory<RowData> operator;
+ FlinkJoinType joinType = joinSpec.getJoinType();
HashJoinType hashJoinType =
HashJoinType.of(
leftIsBuild,
@@ -185,7 +171,7 @@ public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExe
joinType.isRightOuter(),
joinType == FlinkJoinType.SEMI,
joinType == FlinkJoinType.ANTI);
- if (LongHashJoinGenerator.support(hashJoinType, keyType, filterNulls)) {
+ if (LongHashJoinGenerator.support(hashJoinType, keyType, joinSpec.getFilterNulls())) {
operator =
LongHashJoinGenerator.gen(
config,
@@ -206,7 +192,7 @@ public class BatchExecHashJoin extends ExecNodeBase<RowData> implements BatchExe
hashJoinType,
condFunc,
reverseJoin,
- filterNulls,
+ joinSpec.getFilterNulls(),
buildProj,
probeProj,
tryDistinctBuildRow,
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecNestedLoopJoin.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecNestedLoopJoin.java
index fa88df1..6dbea06 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecNestedLoopJoin.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecNestedLoopJoin.java
@@ -37,7 +37,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.calcite.rex.RexNode;
-import java.util.List;
+import java.util.Arrays;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -55,10 +55,11 @@ public class BatchExecNestedLoopJoin extends ExecNodeBase<RowData>
RexNode condition,
boolean leftIsBuild,
boolean singleRowJoin,
- List<ExecEdge> inputEdges,
+ ExecEdge leftEdge,
+ ExecEdge rightEdge,
RowType outputType,
String description) {
- super(inputEdges, outputType, description);
+ super(Arrays.asList(leftEdge, rightEdge), outputType, description);
this.joinType = checkNotNull(joinType);
this.condition = checkNotNull(condition);
this.leftIsBuild = leftIsBuild;
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSortMergeJoin.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSortMergeJoin.java
index 3b3e8fa..465f77e 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSortMergeJoin.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSortMergeJoin.java
@@ -46,7 +46,7 @@ import org.apache.calcite.rex.RexNode;
import javax.annotation.Nullable;
-import java.util.List;
+import java.util.Arrays;
import java.util.stream.IntStream;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -70,10 +70,11 @@ public class BatchExecSortMergeJoin extends ExecNodeBase<RowData>
boolean[] filterNulls,
@Nullable RexNode nonEquiCondition,
boolean leftIsSmaller,
- List<ExecEdge> inputEdges,
+ ExecEdge leftEdge,
+ ExecEdge rightEdge,
RowType outputType,
String description) {
- super(inputEdges, outputType, description);
+ super(Arrays.asList(leftEdge, rightEdge), outputType, description);
this.joinType = checkNotNull(joinType);
this.leftKeys = checkNotNull(leftKeys);
this.rightKeys = checkNotNull(rightKeys);
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalJoin.scala
index 3771979..b9c49a1 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalJoin.scala
@@ -24,17 +24,14 @@ import org.apache.flink.table.planner.plan.nodes.physical.FlinkPhysicalRel
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
import org.apache.flink.table.planner.plan.utils.{JoinTypeUtil, JoinUtil}
-import org.apache.flink.table.runtime.operators.join.FlinkJoinType
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
-import org.apache.calcite.rel.core.{CorrelationId, Join, JoinInfo, JoinRelType}
+import org.apache.calcite.rel.core.{CorrelationId, Join, JoinRelType}
import org.apache.calcite.rel.{RelNode, RelWriter}
import org.apache.calcite.rex.RexNode
import org.apache.calcite.sql.validate.SqlValidatorUtil
-import org.apache.calcite.util.mapping.IntPair
-import java.util
import java.util.Collections
import scala.collection.JavaConversions._
@@ -61,19 +58,6 @@ abstract class CommonPhysicalJoin(
lazy val joinSpec: JoinSpec = JoinUtil.createJoinSpec(this)
- def getJoinInfo: JoinInfo = joinInfo
-
- lazy val filterNulls: Array[Boolean] = {
- val filterNulls = new util.ArrayList[java.lang.Boolean]
- JoinUtil.createJoinInfo(getLeft, getRight, getCondition, filterNulls)
- filterNulls.map(_.booleanValue()).toArray
- }
-
- lazy val keyPairs: List[IntPair] = getJoinInfo.pairs.toList
-
- // TODO remove FlinkJoinType
- lazy val flinkJoinType: FlinkJoinType = JoinTypeUtil.getFlinkJoinType(this.getJoinType)
-
lazy val inputRowType: RelDataType = joinType match {
case JoinRelType.SEMI | JoinRelType.ANTI =>
// Combines inputs' RowType, the result is different from SEMI/ANTI Join's RowType.
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashJoin.scala
index cf96dc2..7659efc 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashJoin.scala
@@ -24,16 +24,16 @@ import org.apache.flink.table.planner.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashJoin
-import org.apache.flink.table.planner.plan.utils.{FlinkRelMdUtil, JoinTypeUtil, JoinUtil}
+import org.apache.flink.table.planner.plan.utils.{FlinkRelMdUtil, JoinUtil}
import org.apache.flink.table.runtime.operators.join.HashJoinType
import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer
+
import org.apache.calcite.plan._
import org.apache.calcite.rel.core._
import org.apache.calcite.rel.metadata.RelMetadataQuery
import org.apache.calcite.rel.{RelNode, RelWriter}
import org.apache.calcite.rex.RexNode
import org.apache.calcite.util.Util
-import java.util
import scala.collection.JavaConversions._
@@ -54,14 +54,6 @@ class BatchPhysicalHashJoin(
val tryDistinctBuildRow: Boolean)
extends BatchPhysicalJoinBase(cluster, traitSet, leftRel, rightRel, condition, joinType) {
- private val (leftKeys, rightKeys) =
- JoinUtil.checkAndGetJoinKeys(keyPairs, getLeft, getRight, allowEmptyKey = true)
- val (buildKeys, probeKeys) = if (leftIsBuild) (leftKeys, rightKeys) else (rightKeys, leftKeys)
-
- // Inputs could be changed. See [[BiRel.replaceInput]].
- def buildRel: RelNode = if (leftIsBuild) getLeft else getRight
- def probeRel: RelNode = if (leftIsBuild) getRight else getLeft
-
val hashJoinType: HashJoinType = HashJoinType.of(
leftIsBuild,
getJoinType.generatesNullsOnRight(),
@@ -162,38 +154,40 @@ class BatchPhysicalHashJoin(
}
override def translateToExecNode(): ExecNode[_] = {
- val nonEquiPredicates = if (!joinInfo.isEqui) {
- joinInfo.getRemaining(getCluster.getRexBuilder)
- } else {
- null
- }
+ JoinUtil.validateJoinSpec(
+ joinSpec,
+ FlinkTypeFactory.toLogicalRowType(left.getRowType),
+ FlinkTypeFactory.toLogicalRowType(right.getRowType))
+
val mq = getCluster.getMetadataQuery
val leftRowSize = Util.first(mq.getAverageRowSize(left), 24).toInt
val leftRowCount = Util.first(mq.getRowCount(left), 200000).toLong
val rightRowSize = Util.first(mq.getAverageRowSize(right), 24).toInt
val rightRowCount = Util.first(mq.getRowCount(right), 200000).toLong
+ val (leftEdge, rightEdge) = getInputEdges
new BatchExecHashJoin(
- JoinTypeUtil.getFlinkJoinType(joinType),
- leftKeys,
- rightKeys,
- filterNulls,
- nonEquiPredicates,
+ joinSpec,
leftRowSize,
rightRowSize,
leftRowCount,
rightRowCount,
leftIsBuild,
tryDistinctBuildRow,
- getInputEdges,
+ leftEdge,
+ rightEdge,
FlinkTypeFactory.toLogicalRowType(getRowType),
getRelDetailedDescription
)
}
- private def getInputEdges: util.List[ExecEdge] = {
+ private def getInputEdges: (ExecEdge, ExecEdge) = {
+
val (buildRequiredShuffle, probeRequiredShuffle) = if (isBroadcast) {
(ExecEdge.RequiredShuffle.broadcast(), ExecEdge.RequiredShuffle.any())
} else {
+ val leftKeys = joinSpec.getLeftKeys
+ val rightKeys = joinSpec.getRightKeys
+ val (buildKeys, probeKeys) = if (leftIsBuild) (leftKeys, rightKeys) else (rightKeys, leftKeys)
(ExecEdge.RequiredShuffle.hash(buildKeys), ExecEdge.RequiredShuffle.hash(probeKeys))
}
val probeDamBehavior = if (hashJoinType.buildLeftSemiOrAnti()) {
@@ -213,9 +207,9 @@ class BatchPhysicalHashJoin(
.build()
if (leftIsBuild) {
- List(buildEdge, probeEdge)
+ (buildEdge, probeEdge)
} else {
- List(probeEdge, buildEdge)
+ (probeEdge, buildEdge)
}
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalNestedLoopJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalNestedLoopJoin.scala
index eac0265..0649878 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalNestedLoopJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalNestedLoopJoin.scala
@@ -118,18 +118,20 @@ class BatchPhysicalNestedLoopJoin(
//~ ExecNode methods -----------------------------------------------------------
override def translateToExecNode(): ExecNode[_] = {
+ val (leftEdge, rightEdge) = getInputEdges
new BatchExecNestedLoopJoin(
JoinTypeUtil.getFlinkJoinType(joinType),
condition,
leftIsBuild,
singleRowJoin,
- getInputEdges,
+ leftEdge,
+ rightEdge,
FlinkTypeFactory.toLogicalRowType(getRowType),
getRelDetailedDescription
)
}
- def getInputEdges: util.List[ExecEdge] = {
+ def getInputEdges: (ExecEdge, ExecEdge) = {
// this is in sync with BatchExecNestedLoopJoinRuleBase#createNestedLoopJoin
val (buildRequiredShuffle, probeRequiredShuffle) = if (joinType == JoinRelType.FULL) {
(ExecEdge.RequiredShuffle.singleton(), ExecEdge.RequiredShuffle.singleton())
@@ -148,9 +150,9 @@ class BatchPhysicalNestedLoopJoin(
.build()
if (leftIsBuild) {
- List(buildEdge, probeEdge)
+ (buildEdge, probeEdge)
} else {
- List(probeEdge, buildEdge)
+ (probeEdge, buildEdge)
}
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortMergeJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortMergeJoin.scala
index 957fc7b..6228fe9 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortMergeJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortMergeJoin.scala
@@ -52,9 +52,6 @@ class BatchPhysicalSortMergeJoin(
val rightSorted: Boolean)
extends BatchPhysicalJoinBase(cluster, traitSet, leftRel, rightRel, condition, joinType) {
- protected lazy val (leftAllKey, rightAllKey) =
- JoinUtil.checkAndGetJoinKeys(keyPairs, getLeft, getRight)
-
protected def isMergeJoinSupportedType(joinRelType: FlinkJoinType): Boolean = {
joinRelType == FlinkJoinType.INNER ||
joinRelType == FlinkJoinType.LEFT ||
@@ -183,14 +180,25 @@ class BatchPhysicalSortMergeJoin(
.build())
override def translateToExecNode(): ExecNode[_] = {
+
+ JoinUtil.validateJoinSpec(
+ joinSpec,
+ FlinkTypeFactory.toLogicalRowType(left.getRowType),
+ FlinkTypeFactory.toLogicalRowType(right.getRowType))
+
new BatchExecSortMergeJoin(
JoinTypeUtil.getFlinkJoinType(joinType),
- leftAllKey,
- rightAllKey,
- filterNulls,
+ joinSpec.getLeftKeys,
+ joinSpec.getRightKeys,
+ joinSpec.getFilterNulls,
condition,
estimateOutputSize(getLeft) < estimateOutputSize(getRight),
- getInputEdges,
+ ExecEdge.builder()
+ .damBehavior(ExecEdge.DamBehavior.END_INPUT)
+ .build(),
+ ExecEdge.builder()
+ .damBehavior(ExecEdge.DamBehavior.END_INPUT)
+ .build(),
FlinkTypeFactory.toLogicalRowType(getRowType),
getRelDetailedDescription
)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecJoin.scala
index 140e333..4a94b94 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecJoin.scala
@@ -29,6 +29,7 @@ import org.apache.flink.table.planner.plan.utils.{JoinUtil, KeySelectorUtil}
import org.apache.flink.table.runtime.operators.join.stream.state.JoinInputSideSpec
import org.apache.flink.table.runtime.operators.join.stream.{StreamingJoinOperator, StreamingSemiAntiJoinOperator}
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
+import org.apache.flink.table.types.logical.RowType
import org.apache.calcite.plan._
import org.apache.calcite.rel.core.{Join, JoinRelType}
@@ -75,13 +76,7 @@ class StreamExecJoin(
val input = getInput(inputOrdinal)
val inputUniqueKeys = getCluster.getMetadataQuery.getUniqueKeys(input)
if (inputUniqueKeys != null) {
- val joinKeys = if (inputOrdinal == 0) {
- // left input
- keyPairs.map(_.source).toArray
- } else {
- // right input
- keyPairs.map(_.target).toArray
- }
+ val joinKeys = if (inputOrdinal == 0) joinSpec.getLeftKeys else joinSpec.getRightKeys
inputUniqueKeys.exists {
uniqueKey => joinKeys.forall(uniqueKey.toArray.contains(_))
}
@@ -130,8 +125,13 @@ class StreamExecJoin(
val leftType = leftTransform.getOutputType.asInstanceOf[InternalTypeInfo[RowData]]
val rightType = rightTransform.getOutputType.asInstanceOf[InternalTypeInfo[RowData]]
- val (leftJoinKey, rightJoinKey) =
- JoinUtil.checkAndGetJoinKeys(keyPairs, getLeft, getRight, allowEmptyKey = true)
+ JoinUtil.validateJoinSpec(
+ joinSpec,
+ getInputNodes.get(0).getOutputType.asInstanceOf[RowType],
+ getInputNodes.get(1).getOutputType.asInstanceOf[RowType],
+ allowEmptyKey = true)
+ val leftJoinKey = joinSpec.getLeftKeys
+ val rightJoinKey = joinSpec.getRightKeys
val leftSelect = KeySelectorUtil.getRowDataSelector(leftJoinKey, leftType)
val rightSelect = KeySelectorUtil.getRowDataSelector(rightJoinKey, rightType)
@@ -141,8 +141,7 @@ class StreamExecJoin(
val generatedCondition = JoinUtil.generateConditionFunction(
tableConfig,
- cluster.getRexBuilder,
- getJoinInfo,
+ joinSpec,
leftType.toRowType,
rightType.toRowType)
@@ -156,7 +155,7 @@ class StreamExecJoin(
generatedCondition,
leftInputSpec,
rightInputSpec,
- filterNulls,
+ joinSpec.getFilterNulls,
minRetentionTime)
} else {
val leftIsOuter = joinType == JoinRelType.LEFT || joinType == JoinRelType.FULL
@@ -169,7 +168,7 @@ class StreamExecJoin(
rightInputSpec,
leftIsOuter,
rightIsOuter,
- filterNulls,
+ joinSpec.getFilterNulls,
minRetentionTime)
}
@@ -198,11 +197,7 @@ class StreamExecJoin(
JoinInputSideSpec.withoutUniqueKey()
} else {
val inRowType = InternalTypeInfo.of(FlinkTypeFactory.toLogicalRowType(input.getRowType))
- val joinKeys = if (input == left) {
- keyPairs.map(_.source).toArray
- } else {
- keyPairs.map(_.target).toArray
- }
+ val joinKeys = if (input == left) joinSpec.getLeftKeys else joinSpec.getRightKeys
val uniqueKeysContainedByJoinKey = uniqueKeys
.filter(uk => uk.toArray.forall(joinKeys.contains(_)))
.map(_.toArray)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index 654aec6..0dfbb02 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -248,8 +248,8 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)
val leftKindSet = getModifyKindSet(children.head)
val rightKindSet = getModifyKindSet(children.last)
- val innerOrSemi = join.flinkJoinType == FlinkJoinType.INNER ||
- join.flinkJoinType == FlinkJoinType.SEMI
+ val innerOrSemi = join.joinSpec.getJoinType == FlinkJoinType.INNER ||
+ join.joinSpec.getJoinType == FlinkJoinType.SEMI
val providedTrait = if (innerOrSemi) {
// forward left and right modify operations
new ModifyKindSetTrait(leftKindSet.union(rightKindSet))
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/JoinUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/JoinUtil.scala
index 47c906c..a419803 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/JoinUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/JoinUtil.scala
@@ -28,15 +28,13 @@ import org.apache.flink.table.types.logical.{LogicalType, RowType}
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.{Join, JoinInfo}
-import org.apache.calcite.rex.{RexBuilder, RexNode, RexUtil}
+import org.apache.calcite.rex.{RexNode, RexUtil}
import org.apache.calcite.util.ImmutableIntList
-import org.apache.calcite.util.mapping.IntPair
import java.util
import java.util.Optional
import scala.collection.JavaConversions._
-import scala.collection.mutable
/**
* Util for [[Join]]s.
@@ -92,52 +90,6 @@ object JoinUtil {
}
/**
- * Check and get join left and right keys.
- */
- def checkAndGetJoinKeys(
- keyPairs: List[IntPair],
- left: RelNode,
- right: RelNode,
- allowEmptyKey: Boolean = false): (Array[Int], Array[Int]) = {
- // get the equality keys
- val leftKeys = mutable.ArrayBuffer.empty[Int]
- val rightKeys = mutable.ArrayBuffer.empty[Int]
- if (keyPairs.isEmpty) {
- if (allowEmptyKey) {
- (leftKeys.toArray, rightKeys.toArray)
- } else {
- throw new TableException(
- s"Joins should have at least one equality condition.\n" +
- s"\tleft: ${left.toString}\n\tright: ${right.toString}\n" +
- s"please re-check the join statement and make sure there's " +
- "equality condition for join.")
- }
- } else {
- // at least one equality expression
- val leftFields = left.getRowType.getFieldList
- val rightFields = right.getRowType.getFieldList
-
- keyPairs.foreach { pair =>
- val leftKeyType = leftFields.get(pair.source).getType.getSqlTypeName
- val rightKeyType = rightFields.get(pair.target).getType.getSqlTypeName
-
- // check if keys are compatible
- if (leftKeyType == rightKeyType) {
- // add key pair
- leftKeys += pair.source
- rightKeys += pair.target
- } else {
- throw new TableException(
- s"Join: Equality join predicate on incompatible types. " +
- s"\tLeft: ${left.toString}\n\tright: ${right.toString}\n" +
- "please re-check the join statement.")
- }
- }
- (leftKeys.toArray, rightKeys.toArray)
- }
- }
-
- /**
* Creates a [[JoinInfo]] by analyzing a condition.
*
* <p>NOTES: the functionality of the method is same with [[JoinInfo#of]],
@@ -163,19 +115,6 @@ object JoinUtil {
def generateConditionFunction(
config: TableConfig,
- rexBuilder: RexBuilder,
- joinInfo: JoinInfo,
- leftType: LogicalType,
- rightType: LogicalType): GeneratedJoinCondition = {
- generateConditionFunction(
- config,
- joinInfo.getRemaining(rexBuilder),
- leftType,
- rightType)
- }
-
- def generateConditionFunction(
- config: TableConfig,
joinSpec: JoinSpec,
leftType: LogicalType,
rightType: LogicalType): GeneratedJoinCondition = {