[SPARK-24385][SQL] Resolve self-join condition ambiguity for all BinaryComparisons

Author: mgaido91

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -1000,11 +1000,11 @@ class Dataset[T] private[sql](
     // By the time we get here, since we have already run analysis, all attributes should've been
     // resolved and become AttributeReference.
     val cond = plan.condition.map { _.transform {
-      case catalyst.expressions.EqualTo(a: AttributeReference, b: AttributeReference)
+      case e @ catalyst.expressions.BinaryComparison(a: AttributeReference, b: AttributeReference)
           if a.sameRef(b) =>
-        catalyst.expressions.EqualTo(
+        e.withNewChildren(Seq(
           withPlan(plan.left).resolve(a.name),
-          withPlan(plan.right).resolve(b.name))
+          withPlan(plan.right).resolve(b.name)))
     }}
 
     withPlan {

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

@@ -287,4 +287,16 @@ class DataFrameJoinSuite extends QueryTest with SharedSQLContext {
       dfOne.join(dfTwo, $"a" === $"b", "left").queryExecution.optimizedPlan
     }
   }
+
+  test("SPARK-24385: Resolve ambiguity in self-joins with operators different from EqualsTo") {
+    withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "false") {
+      val df = spark.range(10)
+      // these should not throw any exception
+      df.join(df, df("id") >= df("id")).queryExecution.optimizedPlan
+      df.join(df, df("id") <=> df("id")).queryExecution.optimizedPlan
+      df.join(df, df("id") <= df("id")).queryExecution.optimizedPlan
+      df.join(df, df("id") > df("id")).queryExecution.optimizedPlan
+      df.join(df, df("id") < df("id")).queryExecution.optimizedPlan
+    }
+  }
 }