[FLINK-36361][table] Do not use StringBuilder for EquivalentExprShuttle

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

@@ -439,17 +439,16 @@ object FlinkRexUtil {
   })
 
   private class EquivalentExprShuttle(rexBuilder: RexBuilder) extends RexShuttle {
-    private val equiExprMap = mutable.HashMap[String, RexNode]()
+    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 =>
-          val swapped = swapOperands(call)
-          if (equiExprMap.contains(swapped.toString)) {
-            swapped
+          if (equiExprSet.contains(call)) {
+            swapOperands(call)
           } else {
-            equiExprMap.put(call.toString, call)
+            equiExprSet.add(call)
             call
           }
         case _ => super.visitCall(call)

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(=($0, 1), =($1, 1)), AND(=($0, 2), =($1, 2))), OR(AND(=($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))))], 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(=(1, $0), =(1, $1)), AND(=(2, $0), =(2, $1))), OR(AND(=(2, $3), =(2, $4)), AND(=(1, $3), =(1, $4))))], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
 ]]>
@@ -152,7 +152,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(=($0, 1), =($1, 1)), =($0, 2)), OR(AND(=($3, 2), =($7, 2)), AND(=($4, 2), =($6, 1), =($7, 1))))], 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(=(1, $0), =(1, $1)), =(2, $0)), OR(AND(=(2, $3), =(2, $7)), AND(=(2, $4), =(1, $6), =(1, $7))))], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
    +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], g=[$6], h=[$7])
       +- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
@@ -177,7 +177,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(=($3, 2), =($4, 2)), AND(=($3, 1), =($4, 1))))], joinType=[inner])
++- 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])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
 ]]>

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=[((a1 = 1) OR (a2 = 2))])
+   :  +- Calc(select=[a1, a2], where=[((1 = a1) OR (2 = a2))])
    :     +- LegacyTableSourceScan(table=[[default_catalog, default_database, A, source: [TestTableSource(a1, a2, a3)]]], fields=[a1, a2, a3])
    +- Exchange(distribution=[single])
-      +- Calc(select=[b1, b2], where=[((b1 = 1) OR (b2 = 2))])
+      +- Calc(select=[b1, b2], where=[((1 = b1) OR (2 = b2))])
          +- LegacyTableSourceScan(table=[[default_catalog, default_database, B, source: [TestTableSource(b1, b2, b3)]]], 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])])
 :     +- LegacyTableSourceScan(table=[[default_catalog, default_database, t, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
 +- Exchange(distribution=[single])
-   +- Calc(select=[x, y, z], where=[(y IS NULL OR (x = 1))])
+   +- Calc(select=[x, y, z], where=[(y IS NULL OR (1 = x))])
       +- LegacyTableSourceScan(table=[[default_catalog, default_database, s, source: [TestTableSource(x, y, z)]]], 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=[((a1 = 1) OR (a2 = 2))])
+   :  +- Calc(select=[a1, a2], where=[((1 = a1) OR (2 = a2))])
    :     +- LegacyTableSourceScan(table=[[default_catalog, default_database, A, source: [TestTableSource(a1, a2, a3)]]], 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=[((a = 1) OR (b = 2))])
+:  +- Calc(select=[a, b], where=[((1 = a) OR (2 = b))])
 :     +- DataStreamScan(table=[[default_catalog, default_database, T1]], fields=[a, b, c])
 +- Exchange(distribution=[single])
    +- Calc(select=[d, e], where=[SEARCH(d, Sarg[1, 5])])
@@ -75,7 +75,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 (d = 3))])
+      +- Calc(select=[d, e, f], where=[(f IS NULL OR (3 = d))])
          +- DataStreamScan(table=[[default_catalog, default_database, T2]], fields=[d, e, f])
 ]]>
     </Resource>