Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5327#discussion_r170296184
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/UpdatingPlanChecker.scala
---
@@ -149,45 +148,39 @@ object UpdatingPlanChecker {
}
case j: DataStreamJoin =>
- val joinType = j.getJoinType
- joinType match {
- case JoinRelType.INNER =>
- // get key(s) for inner join
- val lInKeys = visit(j.getLeft)
- val rInKeys = visit(j.getRight)
- if (lInKeys.isEmpty || rInKeys.isEmpty) {
- None
- } else {
- // Output of inner join must have keys if left and right
both contain key(s).
- // Key groups from both side will be merged by join
equi-predicates
- val lInNames: Seq[String] =
j.getLeft.getRowType.getFieldNames
- val rInNames: Seq[String] =
j.getRight.getRowType.getFieldNames
- val joinNames = j.getRowType.getFieldNames
-
- // if right field names equal to left field names, calcite
will rename right
- // field names. For example, T1(pk, a) join T2(pk, b),
calcite will rename T2(pk, b)
- // to T2(pk0, b).
- val rInNamesToJoinNamesMap = rInNames
- .zip(joinNames.subList(lInNames.size, joinNames.length))
- .toMap
+ // get key(s) for inner join
+ val lInKeys = visit(j.getLeft)
+ val rInKeys = visit(j.getRight)
+ if (lInKeys.isEmpty || rInKeys.isEmpty) {
+ None
+ } else {
+ // Output of inner join must have keys if left and right both
contain key(s).
--- End diff --
remove "inner"?
---
