title: Self-join not working as expected with the DataFrame API
body: <p>I am trying to get the latest records from a table using self join. It works using <code>spark-sql</code> but not working using spark <code>DataFrame</code> API.</p>
  
  <p>Can anyone help? Is it a bug?</p>
  
  <p>I am using Spark 2.2.0 in local mode</p>
  
  <p>Creating input <code>DataFrame</code>:</p>
  
  <pre><code>scala&gt; val df3 = spark.sparkContext.parallelize(Array((1,"a",1),(1,"aa",2),(2,"b",2),(2,"bb",5))).toDF("id","value","time")
  df3: org.apache.spark.sql.DataFrame = [id: int, value: string ... 1 more field]    
  
  scala&gt; val df33 = df3
  df33: org.apache.spark.sql.DataFrame = [id: int, value: string ... 1 more field]
  
  scala&gt; df3.show
  +---+-----+----+
  | id|value|time|
  +---+-----+----+
  |  1|    a|   1|
  |  1|   aa|   2|
  |  2|    b|   2|
  |  2|   bb|   5|
  +---+-----+----+
  
  scala&gt; df33.show
  +---+-----+----+
  | id|value|time|
  +---+-----+----+
  |  1|    a|   1|
  |  1|   aa|   2|
  |  2|    b|   2|
  |  2|   bb|   5|
  +---+-----+----+
  </code></pre>
  
  <p>Now performing the join using SQL: <strong>works</strong></p>
  
  <pre><code>scala&gt; spark.sql("select df33.* from df3 join df33 on df3.id = df33.id and df3.time &lt; df33.time").show
  +---+-----+----+
  | id|value|time|
  +---+-----+----+
  |  1|   aa|   2|
  |  2|   bb|   5|
  +---+-----+----+
  </code></pre>
  
  <p>Now performing the join using dataframe API: <strong>doesn't work</strong></p>
  
  <pre><code>scala&gt; df3.join(df33, (df3.col("id") === df33.col("id")) &amp;&amp; (df3.col("time") &lt; df33.col("time")) ).select(df33.col("id"),df33.col("value"),df33.col("time")).show
  +---+-----+----+
  | id|value|time|
  +---+-----+----+
  +---+-----+----+
  </code></pre>
  
  <p>The thing to notice is the explain plans: <strong>blank for the <code>DataFrame</code> API!!</strong></p>
  
  <pre><code>scala&gt; df3.join(df33, (df3.col("id") === df33.col("id")) &amp;&amp; (df3.col("time") &lt; df33.col("time")) ).select(df33.col("id"),df33.col("value"),df33.col("time")).explain
  == Physical Plan ==
  LocalTableScan &lt;empty&gt;, [id#150, value#151, time#152]
  
  scala&gt; spark.sql("select df33.* from df3 join df33 on df3.id = df33.id and df3.time &lt; df33.time").explain
  == Physical Plan ==
  *Project [id#1241, value#1242, time#1243]
  +- *SortMergeJoin [id#150], [id#1241], Inner, (time#152 &lt; time#1243)
     :- *Sort [id#150 ASC NULLS FIRST], false, 0
     :  +- Exchange hashpartitioning(id#150, 200)
     :     +- *Project [_1#146 AS id#150, _3#148 AS time#152]
     :        +- *SerializeFromObject [assertnotnull(input[0, scala.Tuple3, true])._1 AS _1#146, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString,
  assertnotnull(input[0, scala.Tuple3, true])._2, true) AS _2#147, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#148]
     :           +- Scan ExternalRDDScan[obj#145]
     +- *Sort [id#1241 ASC NULLS FIRST], false, 0
        +- Exchange hashpartitioning(id#1241, 200)
           +- *Project [_1#146 AS id#1241, _2#147 AS value#1242, _3#148 AS time#1243]
              +- *SerializeFromObject [assertnotnull(input[0, scala.Tuple3, true])._1 AS _1#146, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString,
  assertnotnull(input[0, scala.Tuple3, true])._2, true) AS _2#147, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#148]
                 +- Scan ExternalRDDScan[obj#145]
  </code></pre>