[CARMEL-6243][FOLLOW-UP] Insert partial aggregate for right side of left anti join (#1100)

Author: wangyum

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/HandleOuterJoinBuildSideSkew.scala

@@ -76,25 +76,28 @@ object HandleOuterJoinBuildSideSkew extends Rule[LogicalPlan]
         val maxBloomFilterEntries = conf.dynamicBloomFilterJoinPruningMaxBloomFilterEntries
 
         if (rightSize.max > Utils.median(rightSize, false) * threshold) {
-          // 1. Insert bloom filter
-          val insertBF = if (left.stats.rowCount.exists(_ <= maxBloomFilterEntries)) {
-            insertPredicate(rightKeys, right, leftKeys, left)
-          } else {
-            right
-          }
-          // TODO: 2. Insert partial aggregate
-          val joinAttrs = join.condition.map(_.references.filter(canEvaluate(_, right)).toSeq)
-            .getOrElse(Nil)
-          val insertPartialAgg =
-            if (joinAttrs.nonEmpty) PartialAggregate(joinAttrs, joinAttrs, insertBF) else insertBF
+          val (innerJoinRight, leftAntiJoinRight) =
+            if (left.stats.rowCount.exists(_ <= maxBloomFilterEntries)) {
+              // 1. Insert bloom filter
+              val insertBF = insertPredicate(rightKeys, right, leftKeys, left)
+              (insertBF, insertBF)
+            } else {
+              // 2. Insert partial aggregate for right side of left anti join
+              val joinAttrs =
+                join.condition.map(_.references.filter(canEvaluate(_, right)).toSeq.distinct)
+                  .getOrElse(Nil)
+              val insertPartialAgg =
+                if (joinAttrs.nonEmpty) PartialAggregate(joinAttrs, joinAttrs, right) else right
+              (right, insertPartialAgg)
+            }
 
           // Should not convert to BHJ
           val joinHint = JoinHint(Some(HintInfo(strategy = Some(NO_BROADCAST_HASH))), None)
           val union = Union(
-            join.copy(right = insertBF, joinType = Inner, hint = joinHint),
+            join.copy(right = innerJoinRight, joinType = Inner, hint = joinHint),
             Project(left.output ++
               right.output.map(name => Alias(Literal(null, name.dataType), name.name)()),
-              Join(left, insertBF, LeftAnti, join.condition, join.hint)))
+              Join(left, leftAntiJoinRight, LeftAnti, join.condition, join.hint)))
           union.addOptimizeTag(s"created by ${this.simpleRuleName}")
           union
         } else {