title: Spark RAPIDS - Operation not replaced with GPU version
body: <p>I am new to Rapids and I have trouble understanding the supported operations.</p>
  <p>I have data in following format:</p>
  <pre><code>+------------+----------+
  |        kmer|source_seq|
  +------------+----------+
  |TGTCGGTTTAA$|         4|
  |ACCACCACCAC$|         8|
  |GCATAATTTCC$|         1|
  |CCGTCAAAGCG$|         7|
  |CCGTCCCGTGG$|         6|
  |GCGCTGTTATG$|         2|
  |GAGCATAGGTG$|         5|
  |CGGCGGATTCT$|         0|
  |GGCGCGAGGGT$|         3|
  |CCACCACCAC$A|         8|
  |CACCACCAC$AA|         8|
  |CCCAAAAAAAAA|         0|
  |AAGAAAAAAAAA|         5|
  |AAGAAAAAAAAA|         0|
  |TGTAAAAAAAAA|         0|
  |CCACAAAAAAAA|         8|
  |AGACAAAAAAAA|         7|
  |CCCCAAAAAAAA|         0|
  |CAAGAAAAAAAA|         5|
  |TAAGAAAAAAAA|         0|
  +------------+----------+
  </code></pre>
  <p>And to I am trying to find out which &quot;kmer&quot;s have which &quot;source_seq&quot;'s, using the following code:</p>
  <pre><code>val w = Window.partitionBy(&quot;kmer&quot;)
  x.withColumn(&quot;source_seqs&quot;, collect_list(&quot;source_seq&quot;).over(w))
  
  // Result is something like this:
  +------------+----------+-----------+                                           
  |        kmer|source_seq|source_seqs|
  +------------+----------+-----------+
  |AAAACAAGACCA|         2|        [2]|
  |AAAACAAGCAGC|         4|        [4]|
  |AAAACCACGAGC|         3|        [3]|
  |AAAACCGCCAAA|         7|        [7]|
  |AAAACCGGTGTG|         1|        [1]|
  |AAAACCTATATC|         5|        [5]|
  |AAAACGACTTCT|         6|        [6]|
  |AAAACGCGCAAG|         3|        [3]|
  |AAAAGGCCTATT|         7|        [7]|
  |AAAAGGCGTTCG|         3|        [3]|
  |AAAAGGCTGTGA|         1|        [1]|
  |AAAAGGTCTACC|         2|        [2]|
  |AAAAGTCGAGCA|         7|     [7, 0]|
  |AAAAGTCGAGCA|         0|     [7, 0]|
  |AAAATCCGATCA|         0|        [0]|
  |AAAATCGAGCGG|         0|        [0]|
  |AAAATCGTTGAA|         7|        [7]|
  |AAAATGGACAAG|         1|        [1]|
  |AAAATTGCACCA|         3|        [3]|
  |AAACACCGCCGT|         3|        [3]|
  +------------+----------+-----------+
  
  </code></pre>
  <p><a href="https://nvidia.github.io/spark-rapids/docs/supported_ops.html" rel="nofollow noreferrer">The Spark Rapids supported operators documentation</a> mentions <code>collect_list</code> being supported only by windowing, which is what I am doing in my code as far as I know.</p>
  <p>However, looking at the query plan, it is easy to see that the <code>collect_list</code> is not executed by the GPU:</p>
  <pre><code>scala&gt; x.withColumn(&quot;source_seqs&quot;, collect_list(&quot;source_seq&quot;).over(w)).explain
  == Physical Plan ==
  Window [collect_list(source_seq#302L, 0, 0) windowspecdefinition(kmer#301, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS max_source#658], [kmer#301]
  +- GpuColumnarToRow false
     +- GpuSort [kmer#301 ASC NULLS FIRST], false, RequireSingleBatch, 0
        +- GpuCoalesceBatches RequireSingleBatch
           +- GpuShuffleCoalesce 2147483647
              +- GpuColumnarExchange gpuhashpartitioning(kmer#301, 200), ENSURE_REQUIREMENTS, [id=#1496]
                 +- GpuFileGpuScan csv [kmer#301,source_seq#302L] Batched: true, DataFilters: [], Format: CSV, Location: InMemoryFileIndex[file:/home/cloud-user/phase1/example/1620833755/part-00000], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;kmer:string,source_seq:bigint&gt;
  
  </code></pre>
  <p>Unlike a similar query with different function, where we can see the windowing executed with GPU:</p>
  <pre><code>scala&gt; x.withColumn(&quot;min_source&quot;, min(&quot;source_seq&quot;).over(w)).explain
  == Physical Plan ==
  GpuColumnarToRow false
  +- GpuWindow [gpumin(source_seq#302L) gpuwindowspecdefinition(kmer#301, gpuspecifiedwindowframe(RowFrame, gpuspecialframeboundary(unboundedpreceding$()), gpuspecialframeboundary(unboundedfollowing$()))) AS max_source#648L], [kmer#301], false
     +- GpuSort [kmer#301 ASC NULLS FIRST], false, RequireSingleBatch, 0
        +- GpuCoalesceBatches RequireSingleBatch
           +- GpuShuffleCoalesce 2147483647
              +- GpuColumnarExchange gpuhashpartitioning(kmer#301, 200), ENSURE_REQUIREMENTS, [id=#1431]
                 +- GpuFileGpuScan csv [kmer#301,source_seq#302L] Batched: true, DataFilters: [], Format: CSV, Location: InMemoryFileIndex[file:/home/cloud-user/phase1/example/1620833755/part-00000], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;kmer:string,source_seq:bigint&gt;
  </code></pre>
  <p>Am I understanding the supported operations documentation wrong somehow, or have I written the code in a wrong way? Any help for this would be appreciated.</p>