[FLINK-38752][table] Drop EquivalentExprShuttle

Author: snuyanzin

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

@@ -37,7 +37,6 @@ import org.apache.calcite.rex._
 import org.apache.calcite.sql.`type`.SqlTypeName
 import org.apache.calcite.sql.{SqlAsOperator, SqlKind, SqlOperator}
 import org.apache.calcite.sql.fun.{SqlCastFunction, SqlStdOperatorTable}
-import org.apache.calcite.sql.fun.SqlStdOperatorTable._
 import org.apache.calcite.util._
 
 import java.lang.{Iterable => JIterable}
@@ -47,7 +46,6 @@ import java.util.Optional
 import java.util.function.Predicate
 
 import scala.collection.JavaConversions._
-import scala.collection.mutable
 
 /** Utility methods concerning [[RexNode]]. */
 object FlinkRexUtil {
@@ -201,119 +199,14 @@ object FlinkRexUtil {
       RexUtil.composeDisjunction(rexBuilder, nodes)
   }
 
-  /**
-   * Merges same expressions and then simplifies the result expression by [[RexSimplify]].
-   *
-   * Examples for merging same expressions:
-   *   1. a = b AND b = a -> a = b 2. a = b OR b = a -> a = b 3. (a > b AND c < 10) AND b < a -> a >
-   *      b AND c < 10 4. (a > b OR c < 10) OR b < a -> a > b OR c < 10 5. a = a, a >= a, a <= a ->
-   *      true 6. a <> a, a > a, a < a -> false
-   */
+  /** Simplifies the result expression by [[RexSimplify]]. */
   def simplify(rexBuilder: RexBuilder, expr: RexNode, executor: RexExecutor): RexNode = {
     if (expr.isAlwaysTrue || expr.isAlwaysFalse) {
       return expr
     }
 
-    val exprShuttle = new EquivalentExprShuttle(rexBuilder)
-    val equiExpr = expr.accept(exprShuttle)
-    val exprMerger = new SameExprMerger(rexBuilder)
-    val sameExprMerged = exprMerger.mergeSameExpr(equiExpr)
-    val binaryComparisonExprReduced =
-      sameExprMerged.accept(new BinaryComparisonExprReducer(rexBuilder))
-
     val rexSimplify = new RexSimplify(rexBuilder, RelOptPredicateList.EMPTY, executor)
-    rexSimplify.simplifyUnknownAs(binaryComparisonExprReduced, RexUnknownAs.falseIf(true))
-  }
-
-  val BINARY_COMPARISON: util.Set[SqlKind] = util.EnumSet.of(
-    SqlKind.EQUALS,
-    SqlKind.NOT_EQUALS,
-    SqlKind.GREATER_THAN,
-    SqlKind.GREATER_THAN_OR_EQUAL,
-    SqlKind.LESS_THAN,
-    SqlKind.LESS_THAN_OR_EQUAL)
-
-  private class BinaryComparisonExprReducer(rexBuilder: RexBuilder) extends RexShuttle {
-    override def visitCall(call: RexCall): RexNode = {
-      val kind = call.getOperator.getKind
-      if (!kind.belongsTo(BINARY_COMPARISON)) {
-        super.visitCall(call)
-      } else {
-        val operand0 = call.getOperands.get(0)
-        val operand1 = call.getOperands.get(1)
-        (operand0, operand1) match {
-          case (op0: RexInputRef, op1: RexInputRef) if op0.getIndex == op1.getIndex =>
-            kind match {
-              case SqlKind.EQUALS | SqlKind.LESS_THAN_OR_EQUAL | SqlKind.GREATER_THAN_OR_EQUAL =>
-                rexBuilder.makeLiteral(true)
-              case SqlKind.NOT_EQUALS | SqlKind.LESS_THAN | SqlKind.GREATER_THAN =>
-                rexBuilder.makeLiteral(false)
-              case _ => super.visitCall(call)
-            }
-          case _ => super.visitCall(call)
-        }
-      }
-    }
-  }
-
-  private class SameExprMerger(rexBuilder: RexBuilder) extends RexShuttle {
-    private val sameExprMap = mutable.HashMap[String, RexNode]()
-
-    private def mergeSameExpr(expr: RexNode, equiExpr: RexLiteral): RexNode = {
-      if (sameExprMap.contains(expr.toString)) {
-        equiExpr
-      } else {
-        sameExprMap.put(expr.toString, expr)
-        expr
-      }
-    }
-
-    def mergeSameExpr(expr: RexNode): RexNode = {
-      // merges same expressions in the operands of AND and OR
-      // e.g. a = b AND a = b -> a = b AND true
-      //      a = b OR a = b -> a = b OR false
-      val newExpr1 = expr.accept(this)
-
-      // merges same expressions in conjunctions
-      // e.g. (a > b AND c < 10) AND a > b -> a > b AND c < 10 AND true
-      sameExprMap.clear()
-      val newConjunctions = RelOptUtil.conjunctions(newExpr1).map {
-        ex => mergeSameExpr(ex, rexBuilder.makeLiteral(true))
-      }
-      val newExpr2 = newConjunctions.size match {
-        case 0 => newExpr1 // true AND true
-        case 1 => newConjunctions.head
-        case _ => rexBuilder.makeCall(AND, newConjunctions: _*)
-      }
-
-      // merges same expressions in disjunctions
-      // e.g. (a > b OR c < 10) OR a > b -> a > b OR c < 10 OR false
-      sameExprMap.clear()
-      val newDisjunctions = RelOptUtil.disjunctions(newExpr2).map {
-        ex => mergeSameExpr(ex, rexBuilder.makeLiteral(false))
-      }
-      val newExpr3 = newDisjunctions.size match {
-        case 0 => newExpr2 // false OR false
-        case 1 => newDisjunctions.head
-        case _ => rexBuilder.makeCall(OR, newDisjunctions: _*)
-      }
-      newExpr3
-    }
-
-    override def visitCall(call: RexCall): RexNode = {
-      val newCall = call.getOperator match {
-        case AND | OR =>
-          sameExprMap.clear()
-          val newOperands = call.getOperands.map {
-            op =>
-              val value = if (call.getOperator == AND) true else false
-              mergeSameExpr(op, rexBuilder.makeLiteral(value))
-          }
-          call.clone(call.getType, newOperands)
-        case _ => call
-      }
-      super.visitCall(newCall)
-    }
+    rexSimplify.simplifyUnknownAs(expr, RexUnknownAs.falseIf(true))
   }
 
   /**
@@ -412,37 +305,6 @@ object FlinkRexUtil {
       }
     })
 
-  private class EquivalentExprShuttle(rexBuilder: RexBuilder) extends RexShuttle {
-    private val equiExprSet = mutable.HashSet[RexNode]()
-
-    override def visitCall(call: RexCall): RexNode = {
-      call.getOperator match {
-        case EQUALS | NOT_EQUALS | GREATER_THAN | LESS_THAN | GREATER_THAN_OR_EQUAL |
-            LESS_THAN_OR_EQUAL =>
-          if (equiExprSet.contains(call)) {
-            swapOperands(call)
-          } else {
-            equiExprSet.add(call)
-            call
-          }
-        case _ => super.visitCall(call)
-      }
-    }
-
-    private def swapOperands(call: RexCall): RexCall = {
-      val newOp = call.getOperator match {
-        case EQUALS | NOT_EQUALS => call.getOperator
-        case GREATER_THAN => LESS_THAN
-        case GREATER_THAN_OR_EQUAL => LESS_THAN_OR_EQUAL
-        case LESS_THAN => GREATER_THAN
-        case LESS_THAN_OR_EQUAL => GREATER_THAN_OR_EQUAL
-        case _ => throw new IllegalArgumentException(s"Unsupported operator: ${call.getOperator}")
-      }
-      val operands = call.getOperands
-      rexBuilder.makeCall(newOp, operands.last, operands.head).asInstanceOf[RexCall]
-    }
-  }
-
   def getExpressionString(expr: RexNode, inFields: Seq[String]): String = {
     getExpressionString(expr, inFields, ExpressionDetail.Digest)
   }

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml

@@ -98,7 +98,7 @@ LogicalProject(a=[$0], d=[$3])
     <Resource name="optimized rel plan">
       <![CDATA[
 LogicalProject(a=[$0], d=[$3])
-+- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($4, 2)), AND(=($0, 2), =($1, 2), =($3, 1), =($4, 1))), OR(AND(=(1, $0), =(1, $1)), AND(=(2, $0), =(2, $1))), OR(AND(=(2, $3), =(2, $4)), AND(=(1, $3), =(1, $4))))], joinType=[inner])
++- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($4, 2)), AND(=($0, 2), =($1, 2), =($3, 1), =($4, 1))), OR(AND(=($0, 1), =($1, 1)), AND(=($0, 2), =($1, 2))), OR(AND(=($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))))], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 ]]>
@@ -130,7 +130,7 @@ LogicalProject(a=[$0], d=[$6])
     <Resource name="optimized rel plan">
       <![CDATA[
 LogicalProject(a=[$0], d=[$6])
-+- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($7, 2)), AND(=($0, 2), =($4, 2), =($6, 1), =($7, 1))), OR(AND(=(1, $0), =(1, $1)), =(2, $0)), OR(AND(=(2, $3), =(2, $7)), AND(=(2, $4), =(1, $6), =(1, $7))))], joinType=[inner])
++- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($7, 2)), AND(=($0, 2), =($4, 2), =($6, 1), =($7, 1))), OR(AND(=($0, 1), =($1, 1)), =($0, 2)), OR(AND(=($3, 2), =($7, 2)), AND(=($4, 2), =($6, 1), =($7, 1))))], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
    +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], g=[$6], h=[$7])
       +- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
@@ -155,7 +155,7 @@ LogicalProject(a=[$0], d=[$3])
     <Resource name="optimized rel plan">
       <![CDATA[
 LogicalProject(a=[$0], d=[$3])
-+- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))), OR(AND(=(2, $3), =(2, $4)), AND(=(1, $3), =(1, $4))))], joinType=[inner])
++- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($1, 1), =($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))), OR(AND(=($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))))], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
 ]]>

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/JoinTest.xml

@@ -33,10 +33,10 @@ LogicalProject(a1=[$0], b1=[$3])
 Calc(select=[a1, b1])
 +- Join(joinType=[InnerJoin], where=[(((a1 = 1) AND (b1 = 1)) OR ((a2 = 2) AND (b2 = 2)))], select=[a1, a2, b1, b2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
    :- Exchange(distribution=[single])
-   :  +- Calc(select=[a1, a2], where=[((1 = a1) OR (2 = a2))])
+   :  +- Calc(select=[a1, a2], where=[((a1 = 1) OR (a2 = 2))])
    :     +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
    +- Exchange(distribution=[single])
-      +- Calc(select=[b1, b2], where=[((1 = b1) OR (2 = b2))])
+      +- Calc(select=[b1, b2], where=[((b1 = 1) OR (b2 = 2))])
          +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
 ]]>
     </Resource>
@@ -60,7 +60,7 @@ Join(joinType=[InnerJoin], where=[(((a = 1) AND (x = 1)) OR ((a = 2) AND y IS NU
 :  +- Calc(select=[a, b, c], where=[SEARCH(a, Sarg[1, 2])])
 :     +- TableSourceScan(table=[[default_catalog, default_database, t]], fields=[a, b, c])
 +- Exchange(distribution=[single])
-   +- Calc(select=[x, y, z], where=[(y IS NULL OR (1 = x))])
+   +- Calc(select=[x, y, z], where=[(y IS NULL OR (x = 1))])
       +- TableSourceScan(table=[[default_catalog, default_database, s]], fields=[x, y, z])
 ]]>
     </Resource>
@@ -82,7 +82,7 @@ LogicalProject(a1=[$0], b1=[$3])
 Calc(select=[a1, b1])
 +- Join(joinType=[InnerJoin], where=[(((a1 = 1) AND (b1 = 1)) OR (a2 = 2))], select=[a1, a2, b1], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
    :- Exchange(distribution=[single])
-   :  +- Calc(select=[a1, a2], where=[((1 = a1) OR (2 = a2))])
+   :  +- Calc(select=[a1, a2], where=[((a1 = 1) OR (a2 = 2))])
    :     +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
    +- Exchange(distribution=[single])
       +- Calc(select=[b1])

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/JoinTest.xml

@@ -30,7 +30,7 @@ LogicalProject(a=[$0], b=[$1], d=[$3], e=[$4])
       <![CDATA[
 Join(joinType=[InnerJoin], where=[(((a = 1) AND (d = 1)) OR ((b = 2) AND (d = 5)))], select=[a, b, d, e], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
 :- Exchange(distribution=[single])
-:  +- Calc(select=[a, b], where=[((1 = a) OR (2 = b))])
+:  +- Calc(select=[a, b], where=[((a = 1) OR (b = 2))])
 :     +- DataStreamScan(table=[[default_catalog, default_database, T1]], fields=[a, b, c])
 +- Exchange(distribution=[single])
    +- Calc(select=[d, e], where=[SEARCH(d, Sarg[1, 5])])
@@ -56,7 +56,7 @@ Calc(select=[a, b, d, e])
    :  +- Calc(select=[a, b], where=[SEARCH(a, Sarg[0, 1])])
    :     +- DataStreamScan(table=[[default_catalog, default_database, T1]], fields=[a, b, c])
    +- Exchange(distribution=[single])
-      +- Calc(select=[d, e, f], where=[(f IS NULL OR (3 = d))])
+      +- Calc(select=[d, e, f], where=[(f IS NULL OR (d = 3))])
          +- DataStreamScan(table=[[default_catalog, default_database, T2]], fields=[d, e, f])
 ]]>
     </Resource>

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala

@@ -353,8 +353,8 @@ class FlinkRexUtilTest {
             rexBuilder.makeCall(GREATER_THAN, a, rexBuilder.makeLiteral("l"))),
           rexBuilder.makeCall(
             OR,
-            rexBuilder.makeCall(EQUALS, a, b),
-            rexBuilder.makeCall(LESS_THAN, c, d),
+            rexBuilder.makeCall(EQUALS, b, a),
+            rexBuilder.makeCall(GREATER_THAN, d, c),
             rexBuilder.makeCall(LESS_THAN, b, rexBuilder.makeLiteral("k")))
         )
         .toString,