[FEA] Support date cast and date_sub as AST expressions

[viadea]

I wish we can support Left Semi + BNLJ.

eg:

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

val data = Seq(
    Row(Row("Adam ","","Green"),"1","M",1000),
    Row(Row("Bob ","Middle","Green"),"2","M",2000),
    Row(Row("Cathy ","","Green"),"3","F",3000)
)

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))

val df = spark.createDataFrame(spark.sparkContext.parallelize(data),schema)
val df2 = df.withColumn("dt",current_date().as("dt")).withColumn("ts",current_timestamp().as("ts"))

df2.printSchema
df2.write.format("parquet").mode("overwrite").save("/tmp/testparquet")
val newdf2 = spark.read.parquet("/tmp/testparquet")
newdf2.createOrReplaceTempView("df2")


val query="""
select count(*) as cnt 
from df2 a left semi join 
df2 b on date(a.ts) > b.dt - 7  
"""

spark.sql(query).collect

Not-supported-messages:

        !Exec <BroadcastNestedLoopJoinExec> cannot run on GPU because not all expressions can be replaced
          @Expression <GreaterThan> (cast(ts#127 as date) > date_sub(dt#253, 7)) could run on GPU
            !Expression <Cast> cast(ts#127 as date) cannot run on GPU because AST is required and this expression does not support AST
              @Expression <AttributeReference> ts#127 could run on GPU
            !Expression <DateSub> date_sub(dt#253, 7) cannot run on GPU because AST is required and this expression does not support AST
              @Expression <AttributeReference> dt#253 could run on GPU

[jlowe]

This request is more about asking cast-as-date and date_sub to be supported as AST expressions than the join itself. If those were supported by AST then the join would have been on the GPU.

[NVnavkumar]

Note if this issue is resolved, then the integration test used in #7037 will need to be updated as it currently relies on this running on the CPU (an unfortunate limitation of the testing scenario involved).

[revans2]

I think we need to be clear about a lot of what is being asked here. Casting a timestamp to a date is complex, especially if a timezone is involved.

spark-rapids/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCast.scala

Lines 628 to 633 in a056f16

	case (TimestampType, DateType) if options.timeZoneId.isDefined =>
	val zoneId = DateTimeUtils.getZoneId(options.timeZoneId.get)
	withResource(GpuTimeZoneDB.fromUtcTimestampToTimestamp(input.asInstanceOf[ColumnVector],
	zoneId.normalized())) {
	shifted => shifted.castTo(GpuColumnVector.getNonNestedRapidsType(toDataType))
	}

But even without a timezone it is not likely to ever be an AST operation. Date differences, however, should be a subtraction operation.

spark-rapids/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala

Line 1241 in a056f16

override def binaryOp: BinaryOp = BinaryOp.SUB

So putting that in AST should be super simple to do, if we can get CUDF to ignore the logical type and just look at the physical type of the data.

That said I think the simplest thing for us to do to be able to support complex operations is to finish #9635

It added in some basic support for rewriting joins so part of it is on the AST and other parts are not. But it only did it for a BroadcastNestedLoopJoin. We really should just extend that out to all join types and do some performance testing to see how well it performs.

[viadea]

Note: I tested the original query in this issue and it works now on GPU.
I will keep this issue open as discussed.

In the meantime, I opened 2 related FEAs based on a customer's fallback messages:
#11213
#11214