Support multiplication on ANSI interval types

integration_tests/src/main/python/arithmetic_ops_test.py

@@ -867,3 +867,11 @@ def test_subtraction_overflow_with_ansi_enabled(data, tp, expr):
         assert_gpu_and_cpu_are_equal_collect(
             func=lambda spark: _get_overflow_df(spark, data, tp, expr),
             conf=ansi_enabled_conf)
+
+@pytest.mark.skipif(is_before_spark_330(), reason='DayTimeInterval is not supported before Pyspark 3.3.0')
+@pytest.mark.parametrize('data_gen', _no_overflow_multiply_gens + [DoubleGen(min_exp=-3, max_exp=5, special_cases=[0.0])], ids=idfn)
+def test_day_time_interval_multiply_number(data_gen):
+    gen_list = [('_c1', DayTimeIntervalGen(min_value=timedelta(seconds=-20 * 86400), max_value=timedelta(seconds=20 * 86400))),
+                ('_c2', data_gen)]
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: gen_df(spark, gen_list).selectExpr("_c1 * _c2"))

sql-plugin/src/main/320+/scala/com/nvidia/spark/rapids/shims/Spark320PlusShims.scala

@@ -250,7 +250,7 @@ trait Spark320PlusShims extends SparkShims with RebaseShims with Logging {
       ExprChecks.projectAndAst(
         TypeSig.astTypes,
         (TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_128 + TypeSig.CALENDAR
-          + TypeSig.ARRAY + TypeSig.MAP + TypeSig.STRUCT + TypeSig.DAYTIME)
+          + TypeSig.ARRAY + TypeSig.MAP + TypeSig.STRUCT + TypeSig.YEARMONTH + TypeSig.DAYTIME)
           .nested(TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.DECIMAL_128 +
             TypeSig.ARRAY + TypeSig.MAP + TypeSig.STRUCT),
         TypeSig.all),

sql-plugin/src/main/330+/scala/com/nvidia/spark/rapids/shims/GpuTypeShims.scala

@@ -135,6 +135,7 @@ object GpuTypeShims {
    */
   def supportToScalarForType(t: DataType): Boolean = {
     t match {
+      case _: YearMonthIntervalType => true
       case _: DayTimeIntervalType => true
       case _ => false
     }
@@ -143,8 +144,13 @@ object GpuTypeShims {
   /**
    * Convert the given value to Scalar
    */
-  def toScalarForType(t: DataType, v: Any) = {
+  def toScalarForType(t: DataType, v: Any): Scalar = {
     t match {
+      case _: YearMonthIntervalType => v match {
+        case i: Int => Scalar.fromInt(i)
+        case _ => throw new IllegalArgumentException(s"'$v: ${v.getClass}' is not supported" +
+            s" for IntType, expecting int")
+      }
       case _: DayTimeIntervalType => v match {
         case l: Long => Scalar.fromLong(l)
         case _ => throw new IllegalArgumentException(s"'$v: ${v.getClass}' is not supported" +

sql-plugin/src/main/330+/scala/com/nvidia/spark/rapids/shims/Spark33XShims.scala

@@ -34,7 +34,7 @@ import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.rapids._
 import org.apache.spark.sql.rapids.execution.GpuShuffleMeta
-import org.apache.spark.sql.rapids.shims.GpuTimeAdd
+import org.apache.spark.sql.rapids.shims.{GpuMultiplyDTInterval, GpuMultiplyYMInterval, GpuTimeAdd}
 import org.apache.spark.sql.types.{CalendarIntervalType, DayTimeIntervalType, DecimalType, StructType}
 import org.apache.spark.unsafe.types.CalendarInterval
 
@@ -312,6 +312,39 @@ trait Spark33XShims extends Spark321PlusShims with Spark320PlusNonDBShims {
         (a, conf, p, r) => new BinaryAstExprMeta[LessThanOrEqual](a, conf, p, r) {
           override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression =
             GpuLessThanOrEqual(lhs, rhs)
+        }),
+      GpuOverrides.expr[Alias](
+        "Gives a column a name",
+        ExprChecks.unaryProjectAndAstInputMatchesOutput(
+          TypeSig.astTypes,
+          (TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.MAP + TypeSig.ARRAY + TypeSig.STRUCT
+              + TypeSig.DECIMAL_128 + TypeSig.YEARMONTH + TypeSig.DAYTIME).nested(),
+          TypeSig.all),
+        (a, conf, p, r) => new UnaryAstExprMeta[Alias](a, conf, p, r) {
+          override def convertToGpu(child: Expression): GpuExpression =
+            GpuAlias(child, a.name)(a.exprId, a.qualifier, a.explicitMetadata)
+        }),
+      GpuOverrides.expr[MultiplyYMInterval](
+        "Year-month interval * operator",
+        ExprChecks.binaryProject(
+          TypeSig.YEARMONTH,
+          TypeSig.YEARMONTH,
+          ("lhs", TypeSig.YEARMONTH, TypeSig.YEARMONTH),
+          ("rhs", TypeSig.gpuNumeric - TypeSig.DECIMAL_128, TypeSig.gpuNumeric)),
+        (a, conf, p, r) => new BinaryExprMeta[MultiplyYMInterval](a, conf, p, r) {
+          override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression =
+            GpuMultiplyYMInterval(lhs, rhs)
+        }),
+      GpuOverrides.expr[MultiplyDTInterval](
+        "Day-time interval * operator",
+        ExprChecks.binaryProject(
+          TypeSig.DAYTIME,
+          TypeSig.DAYTIME,
+          ("lhs", TypeSig.DAYTIME, TypeSig.DAYTIME),
+          ("rhs", TypeSig.gpuNumeric - TypeSig.DECIMAL_128, TypeSig.gpuNumeric)),
+        (a, conf, p, r) => new BinaryExprMeta[MultiplyDTInterval](a, conf, p, r) {
+          override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression =
+            GpuMultiplyDTInterval(lhs, rhs)
         })
     ).map(r => (r.getClassFor.asSubclass(classOf[Expression]), r)).toMap
     super.getExprs ++ map
@@ -456,7 +489,7 @@ trait Spark33XShims extends Spark321PlusShims with Spark320PlusNonDBShims {
         "The backend for most select, withColumn and dropColumn statements",
         ExecChecks(
           (TypeSig.commonCudfTypes + TypeSig.NULL + TypeSig.STRUCT + TypeSig.MAP +
-              TypeSig.ARRAY + TypeSig.DECIMAL_128 + TypeSig.DAYTIME).nested(),
+              TypeSig.ARRAY + TypeSig.DECIMAL_128 + TypeSig.YEARMONTH + TypeSig.DAYTIME).nested(),
           TypeSig.all),
         (proj, conf, p, r) => new GpuProjectExecMeta(proj, conf, p, r)),
       GpuOverrides.exec[FilterExec](