check if expectedColStats.size == filterStats.attributeStats.size

Author: ron8hu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala

@@ -662,6 +662,7 @@ case class FilterEstimation(plan: Filter, catalystConf: CatalystConf) extends Lo
         // If null values exists, we set it to partial overlap.
         (((maxLeft < minRight) || (maxRight < minLeft)) && allNotNull,
           (minLeft == minRight) && (maxLeft == maxRight) && allNotNull
+            && (colStatLeft.distinctCount == colStatRight.distinctCount)
         )
     }
 
@@ -754,18 +755,16 @@ case class FilterEstimation(plan: Filter, catalystConf: CatalystConf) extends Lo
             //         ^    filtered
             //         |
             //     newMaxLeft
-
-
           if (minLeft < minRight) {
-              newMinLeft = colStatRight.min
-            } else {
-              newMinRight = colStatLeft.min
-            }
-            if (maxLeft < maxRight) {
-              newMaxRight = colStatLeft.max
-            } else {
-              newMaxLeft = colStatRight.max
-            }
+            newMinLeft = colStatRight.min
+          } else {
+            newMinRight = colStatLeft.min
+          }
+          if (maxLeft < maxRight) {
+            newMaxRight = colStatLeft.max
+          } else {
+            newMaxLeft = colStatRight.max
+          }
         }
 
         val newStatsLeft = colStatLeft.copy(distinctCount = newNdvLeft, min = newMinLeft,

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala

@@ -609,6 +609,7 @@ class FilterEstimationSuite extends StatsEstimationTestBase {
       if (rowCountValue != 0) {
         // Need to check attributeStats one by one because we may have multiple output columns.
         // Due to update operation, the output columns may be in different order.
+        assert(expectedColStats.size == filterStats.attributeStats.size)
         expectedColStats.foreach { kv =>
           val filterColumnStat = filterStats.attributeStats.get(kv._1).get
           assert(filterColumnStat == kv._2)