Do not push down join predicate that are ambiguous to both sides

Author: EnricoMi

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

@@ -69,7 +69,10 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan]
       val canPushDownPredicate = (predicate: Expression) => {
         val replaced = replaceAlias(predicate, aliasMap)
         predicate.references.nonEmpty &&
-          replaced.references.subsetOf(agg.child.outputSet ++ rightOp.outputSet)
+          replaced.references.subsetOf(agg.child.outputSet ++ rightOp.outputSet) &&
+          // references must not be ambiguous (i.e., not contained in both sides)
+          !replaced.references.exists(attr =>
+            agg.child.outputSet.contains(attr) && rightOp.outputSet.contains(attr))
       }
       val makeJoinCondition = (predicates: Seq[Expression]) => {
         replaceAlias(predicates.reduce(And), aliasMap)

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

@@ -59,6 +59,13 @@ class DataFrameSuite extends QueryTest
   with AdaptiveSparkPlanHelper {
   import testImplicits._
 
+  test("aggr antijoin") {
+    val ids = Seq(1, 2, 3).toDF("id").distinct()
+    val result = ids.withColumn("id", $"id" + 1)
+      .join(ids, "id", "left_anti").collect()
+    assert(result.length == 1)
+  }
+
   test("analysis error should be eagerly reported") {
     intercept[Exception] { testData.select("nonExistentName") }
     intercept[Exception] {