Add in running window optimization using scan

[revans2]

Spark optimizes running windows to have a linear time algorithm. When discussing this with cudf rapidsai/cudf#8440 it was decided to use scan and segmented_scan (group by scan). This puts in a framework for this and adds in a few initial implementations.

In performance tests on my local box row_number and count are only slightly faster now than they were under window, but min, max, and sum all show significant performance gains similar to those I showed were possible in rapidsai/cudf#8440

In a large max running window with no partition by I have seen performance improvements of 171x faster cold and 542x faster hot compared to the CPU

scala> spark.time(spark.range(0L, Int.MaxValue, 1L, numPartitions=1).select(max("id").over(Window.orderBy("id").rowsBetween(Window.unboundedPreceding, 0)).as("RN"), col("id")).orderBy(desc("RN")).show)
...
Time taken: 3633 ms
scala> spark.time(spark.range(0L, Int.MaxValue, 1L, numPartitions=1).select(max("id").over(Window.orderBy("id").rowsBetween(Window.unboundedPreceding, 0)).as("RN"), col("id")).orderBy(desc("RN")).show)
...
Time taken: 1153 ms
scala> spark.conf.set("spark.rapids.sql.enabled", "false")
scala> spark.time(spark.range(0L, Int.MaxValue, 1L, numPartitions=1).select(max("id").over(Window.orderBy("id").rowsBetween(Window.unboundedPreceding, 0)).as("RN"), col("id")).orderBy(desc("RN")).show)
...
Time taken: 622094 ms
scala> spark.conf.set("spark.rapids.sql.enabled", "false")
scala> spark.time(spark.range(0L, Int.MaxValue, 1L, numPartitions=1).select(max("id").over(Window.orderBy("id").rowsBetween(Window.unboundedPreceding, 0)).as("RN"), col("id")).orderBy(desc("RN")).show)
...
Time taken: 625660 ms

This is a special case because when no partition is given the data all goes to a single task, so it needs a single core to process the data. But as you can see it would still take hundreds of CPU cores in the partitioned cast to offset the performance gains.

On the previous GPU code I could not run this because I had to kill it before my GPU overheated.

This is stepping stone to be able to support rank and dense_rank

[revans2]

I tested this on databricks and it works there too.

[revans2]

build

[revans2]

With the review work I accidentally checked in a change that makes this require the fix from rapidsai/cudf#8705. I am inclined to wait for it to get merged in, but if others want to merge this in sooner I can revert the small change and do a follow on PR when the cudf change does get merged in.

[revans2]

build

[revans2]

build