title: Avoid repartition costs when filtering and then coalescing
body: <p>I am implementing a range query on an RDD of (x,y) points in pyspark. I partitioned the xy space into a 16*16 grid (256 cells) and assigned each point in my RDD to one of these cells.
  The gridMappedRDD is a PairRDD: <code>(cell_id, Point object)</code></p>
  
  <p>I partitioned this RDD to 256 partitions, using:</p>
  
  <pre><code>gridMappedRDD.partitionBy(256)
  </code></pre>
  
  <p>The range query is a rectangular box. I have a method for my Grid object which can return the list of cell ids which overlap with the query range. So, I used this as a filter to prune the unrelated cells:</p>
  
  <pre><code>filteredRDD = gridMappedRDD.filter(lambda x: x[0] in candidateCells)
  </code></pre>
  
  <p>But the problem is that when running the query and then collecting the results, all the 256 partitions are evaluated; A task is created for each partition.</p>
  
  <p>To avoid this problem, I tried coalescing the filteredRDD to the length of candidateCell list and I hoped this could solve the problem.</p>
  
  <pre><code>filteredRDD.coalesce(len(candidateCells))
  </code></pre>
  
  <p>In fact the resulting RDD has <code>len(candidateCells)</code> partitions but the partitions are not the same as <code>gridMappedRDD</code>.</p>
  
  <p>As stated in the coalesce documentation, the <code>shuffle</code> parameter is False and no shuffle should be performed among partitions but I can see (with the help of glom()) that this is not the case.</p>
  
  <p>For example after a <code>coalesce(4)</code> with <code>candidateCells=[62, 63, 78, 79]</code> the partitions are like this:</p>
  
  <pre><code>[[(62, P), (62, P) .... , (63, P)],
   [(78, P), (78, P) .... , (79, P)],
   [], []
  ]
  </code></pre>
  
  <p>Actually, by coalescing, I have a shuffle read which equals to the size of my whole dataset for every task, which takes a significant time. What I need is an RDD with only partitions related to cells in candidateCells, without any shuffles. 
  So, my question is that is it possible to filter only some partitions without reshuffling? For the above example, my filteredRDD would have 4 partitions with exactly the same data as originalRDD's 62, 63, 78, 79th partitions. Doing so, the query could be directed to affecting partitions only.</p>