[FEA] Support last in windowing context for Integer type.

[viadea]

I wish we can support function last in windowing context for IntegerType/DoubleType.

For example:

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val data = Seq(
    Row(Row("Adam ","","Green"),"1","M",1000, "2019-01-01",List("Java","Scala")),
    Row(Row("Bob ","Middle","Green"),"2","M",2000, "2019-01-02",List("Java","Python")),
    Row(Row("Cathy ","","Green"),"3","F",3000, "2019-01-03",List())
)

val schema = (new StructType()
  .add("name",new StructType()
    .add("firstname",StringType)
    .add("middlename",StringType)
    .add("lastname",StringType)) 
  .add("id",StringType)
  .add("gender",StringType)
  .add("salary",IntegerType)
  .add("birthdayStr",StringType)
  .add("language",ArrayType(StringType))
             )

val df = spark.createDataFrame(spark.sparkContext.parallelize(data),schema)
df.withColumn("birthday", to_date(col("birthdayStr"))).write.format("parquet").mode("overwrite").save("/tmp/testparquet")
val df2 = spark.read.parquet("/tmp/testparquet")
df2.createOrReplaceTempView("df2")
df2.printSchema

sql("""SELECT gender, last(salary,true) OVER (PARTITION BY gender ORDER BY salary) FROM df2""").collect

[revans2]

This is essentially the same as #4005

[sameerz]

cc: @mythrocks

[mythrocks]

I have taken myself off the assignee list, leaving this issue in the capable hands of @res-life.

[mythrocks]

@res-life, here is a reference implementation for supporting FIRST() and LAST() in spark-rapids. I haven't included support for NTH_VALUE(), but it should be along these lines.

[res-life]

Note:
Cudf does not support range queries on floats or doubles yet.
Still depends on cuDF: #6000 (comment)

[res-life]

Split this issue into 2, one for Integer, one for Double #6193

[res-life]

#6000 already supported Integer type.