title: Spark is sorting already sorted partitions resulting in performance loss
body: <p>For a cached dataframe, partitioned and sorted within partitions, I get good performance when querying the key with a <code>where</code> clause but bad performance when performing a join with a small table on the same key.</p>
  
  <p>See example dataset <code>dftest</code> below with 10Kx44K = 438M rows.</p>
  
  <pre><code>sqlContext.sql(f'set spark.sql.shuffle.partitions={32}')
  sqlContext.clearCache()
  sc.setCheckpointDir('/checkpoint/temp')
  import datetime
  from pyspark.sql.functions import *
  from pyspark.sql import Row
  
  start_date = datetime.date(1900, 1, 1)
  end_date   = datetime.date(2020, 1, 1)
  
  dates = [ start_date + datetime.timedelta(n) for n in range(int ((end_date - start_date).days))]
  
  dfdates=spark.createDataFrame(list(map(lambda x: Row(date=x), dates))) # some dates
  dfrange=spark.createDataFrame(list(map(lambda x: Row(number=x), range(10000)))) # some number range
  
  dfjoin = dfrange.crossJoin(dfdates)
  dftest = dfjoin.withColumn("random1", round(rand()*(10-5)+5,0)).withColumn("random2", round(rand()*(10-5)+5,0)).withColumn("random3", round(rand()*(10-5)+5,0)).withColumn("random4", round(rand()*(10-5)+5,0)).withColumn("random5", round(rand()*(10-5)+5,0)).checkpoint()
  dftest = dftest.repartition("number").sortWithinPartitions("number", "date").cache()
  dftest.count() # 438,290,000 rows
  </code></pre>
  
  <p>The following query now takes roughly a second (on a small cluster with 2 workers):</p>
  
  <pre><code>dftest.where("number = 1000 and date = \"2001-04-04\"").count()
  </code></pre>
  
  <p>However, when I write a similar condition as a join, it takes 2 minutes:</p>
  
  <pre><code>dfsub = spark.createDataFrame([(10,"1900-01-02",1),
    (1000,"2001-04-04",2),
    (4000,"2002-05-05",3),
    (5000,"1950-06-06",4),
    (9875,"1980-07-07",5)],
  ["number","date", "dummy"]).repartition("number").sortWithinPartitions("number", "date").cache()
  df_result = dftest.join(dfsub, ( dftest.number == dfsub.number ) &amp; ( dftest.date == dfsub.date ), 'inner').cache()
  df_result.count() # takes 2 minutes (result = 5)
  </code></pre>
  
  <p>I would have expected this to be roughly equally fast.  Especially since I would hope that the larger dataframe is already clustered and cached.  Looking at the plan:</p>
  
  <pre><code>== Physical Plan ==
  InMemoryTableScan [number#771L, date#769, random1#775, random2#779, random3#784, random4#790, random5#797, number#945L, date#946, dummy#947L]
     +- InMemoryRelation [number#771L, date#769, random1#775, random2#779, random3#784, random4#790, random5#797, number#945L, date#946, dummy#947L], StorageLevel(disk, memory, deserialized, 1 replicas)
           +- *(3) SortMergeJoin [number#771L, cast(date#769 as string)], [number#945L, date#946], Inner
              :- *(1) Sort [number#771L ASC NULLS FIRST, cast(date#769 as string) ASC NULLS FIRST], false, 0
              :  +- *(1) Filter (isnotnull(number#771L) &amp;&amp; isnotnull(date#769))
              :     +- InMemoryTableScan [number#771L, date#769, random1#775, random2#779, random3#784, random4#790, random5#797], [isnotnull(number#771L), isnotnull(date#769)]
              :           +- InMemoryRelation [number#771L, date#769, random1#775, random2#779, random3#784, random4#790, random5#797], StorageLevel(disk, memory, deserialized, 1 replicas)
              :                 +- Sort [number#771L ASC NULLS FIRST, date#769 ASC NULLS FIRST], false, 0
              :                    +- Exchange hashpartitioning(number#771L, 32)
              :                       +- *(1) Scan ExistingRDD[number#771L,date#769,random1#775,random2#779,random3#784,random4#790,random5#797]
              +- *(2) Filter (isnotnull(number#945L) &amp;&amp; isnotnull(date#946))
                 +- InMemoryTableScan [number#945L, date#946, dummy#947L], [isnotnull(number#945L), isnotnull(date#946)]
                       +- InMemoryRelation [number#945L, date#946, dummy#947L], StorageLevel(disk, memory, deserialized, 1 replicas)
                             +- Sort [number#945L ASC NULLS FIRST, date#946 ASC NULLS FIRST], false, 0
                                +- Exchange hashpartitioning(number#945L, 32)
                                   +- *(1) Scan ExistingRDD[number#945L,date#946,dummy#947L]
  </code></pre>
  
  <p>A lot of time seems to be spent sorting the larger dataframe by number and date (this line: <code>Sort [number#771L ASC NULLS FIRST, date#769 ASC NULLS FIRST], false, 0</code>). It leaves me with the following questions:</p>
  
  <ul>
  <li>within the partitions, the sort order for both the left and right side is exactly the same, and optimal for the JOIN clause, why is Spark still sorting the the partitions again?</li>
  <li>as the 5 join records match (up to) 5 partitions, why are all partitions evaluated?</li>
  <li>It seems Catalyst is not using the info of <code>repartition</code> and <code>sortWithinPartitions</code> of the cached dataframe. Does it make sense to use <code>sortWithinPartitions</code> in cases like these?</li>
  </ul>