feat: support multiply_dt_interval

native/core/src/execution/planner.rs

@@ -362,6 +362,9 @@ impl PhysicalPlanner {
                         DataType::Time64(TimeUnit::Nanosecond) => {
                             ScalarValue::Time64Nanosecond(None)
                         }
+                        DataType::Duration(TimeUnit::Microsecond) => {
+                            ScalarValue::DurationMicrosecond(None)
+                        }
                         dt => {
                             return Err(GeneralError(format!("{dt:?} is not supported in Comet")))
                         }
@@ -388,9 +391,12 @@ impl PhysicalPlanner {
                             DataType::Timestamp(TimeUnit::Microsecond, Some(tz)) => {
                                 ScalarValue::TimestampMicrosecond(Some(*value), Some(tz))
                             }
+                            DataType::Duration(TimeUnit::Microsecond) => {
+                                ScalarValue::DurationMicrosecond(Some(*value))
+                            }
                             dt => {
                                 return Err(GeneralError(format!(
-                                    "Expected either 'Int64' or 'Timestamp' for LongVal, but found {dt:?}"
+                                    "Expected 'Int64', 'Timestamp', or 'Duration(Microsecond)' for LongVal, but found {dt:?}"
                                 )))
                             }
                         },

spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala

@@ -61,6 +61,7 @@ private[codegen] object CometBatchKernelCodegenInput {
     classOf[Float4Vector],
     classOf[Float8Vector],
     classOf[DateDayVector],
+    classOf[DurationVector],
     classOf[TimeStampMicroVector],
     classOf[TimeStampMicroTZVector])
   private val cometPlainVectorName: String = classOf[CometPlainVector].getName
@@ -133,6 +134,7 @@ private[codegen] object CometBatchKernelCodegenInput {
     val longCases = withOrd.collect {
       case (ArrowColumnSpec(cls, _), ord)
           if cls == classOf[BigIntVector] ||
+            cls == classOf[DurationVector] ||
             cls == classOf[TimeStampMicroVector] ||
             cls == classOf[TimeStampMicroTZVector] =>
         s"      case $ord: return this.col$ord.getLong(this.rowIdx);"
@@ -591,7 +593,8 @@ private[codegen] object CometBatchKernelCodegenInput {
     case ByteType => s"getByte($idx)"
     case ShortType => s"getShort($idx)"
     case IntegerType | DateType => s"getInt($idx)"
-    case LongType | TimestampType | TimestampNTZType => s"getLong($idx)"
+    case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
+      s"getLong($idx)"
     case FloatType => s"getFloat($idx)"
     case DoubleType => s"getDouble($idx)"
     case d: DecimalType => s"getDecimal($idx, ${d.precision}, ${d.scale})"
@@ -692,7 +695,7 @@ private[codegen] object CometBatchKernelCodegenInput {
            |      public int getInt(int i) {
            |        return $childField.getInt(startIndex + i);
            |      }""".stripMargin
-      case LongType | TimestampType | TimestampNTZType =>
+      case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
         s"""      @Override
            |      public long getLong(int i) {
            |        return $childField.getLong(startIndex + i);
@@ -845,7 +848,7 @@ private[codegen] object CometBatchKernelCodegenInput {
           s"        case $fi: return ${path}_f$fi.getShort(this.rowIdx);"
         case IntegerType | DateType =>
           s"        case $fi: return ${path}_f$fi.getInt(this.rowIdx);"
-        case LongType | TimestampType | TimestampNTZType =>
+        case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
           s"        case $fi: return ${path}_f$fi.getLong(this.rowIdx);"
         case FloatType =>
           s"        case $fi: return ${path}_f$fi.getFloat(this.rowIdx);"
@@ -897,8 +900,9 @@ private[codegen] object CometBatchKernelCodegenInput {
     val longCases = scalarOrd.collect {
       case (f, fi)
           if f.sparkType == LongType || f.sparkType == TimestampType ||
-            f.sparkType == TimestampNTZType =>
-        fieldReadScalar(fi, LongType, f.nullable)
+            f.sparkType == TimestampNTZType ||
+            f.sparkType.isInstanceOf[DayTimeIntervalType] =>
+        fieldReadScalar(fi, f.sparkType, f.nullable)
     }
     val floatCases =
       scalarOrd.collect {

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -300,6 +300,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[MakeTimestamp] -> CometMakeTimestamp,
       classOf[MakeYMInterval] -> CometMakeYMInterval,
       classOf[MakeDTInterval] -> CometMakeDTInterval,
+      classOf[MultiplyDTInterval] -> CometMultiplyDTInterval,
       classOf[MicrosToTimestamp] -> CometMicrosToTimestamp,
       classOf[MillisToTimestamp] -> CometMillisToTimestamp,
       classOf[MonthsBetween] -> CometMonthsBetween,

spark/src/main/scala/org/apache/comet/serde/datetime.scala

@@ -21,7 +21,7 @@ package org.apache.comet.serde
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.expressions.{AddMonths, Attribute, ConvertTimezone, DateAdd, DateDiff, DateFormatClass, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, Expression, FromUTCTimestamp, GetDateField, GetTimestamp, Hour, Hours, LastDay, Literal, MakeDate, MakeDTInterval, MakeTimestamp, MakeYMInterval, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, NextDay, PreciseTimestampConversion, Quarter, Second, SecondsToTimestamp, ToUnixTimestamp, ToUTCTimestamp, TruncDate, TruncTimestamp, UnixDate, UnixMicros, UnixMillis, UnixSeconds, UnixTimestamp, WeekDay, WeekOfYear, Year}
+import org.apache.spark.sql.catalyst.expressions.{AddMonths, Attribute, ConvertTimezone, DateAdd, DateDiff, DateFormatClass, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, Expression, FromUTCTimestamp, GetDateField, GetTimestamp, Hour, Hours, LastDay, Literal, MakeDate, MakeDTInterval, MakeTimestamp, MakeYMInterval, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, MultiplyDTInterval, NextDay, PreciseTimestampConversion, Quarter, Second, SecondsToTimestamp, ToUnixTimestamp, ToUTCTimestamp, TruncDate, TruncTimestamp, UnixDate, UnixMicros, UnixMillis, UnixSeconds, UnixTimestamp, WeekDay, WeekOfYear, Year}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataType, DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -954,6 +954,8 @@ object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval]
 
 object CometMakeDTInterval extends CometCodegenDispatch[MakeDTInterval]
 
+object CometMultiplyDTInterval extends CometCodegenDispatch[MultiplyDTInterval]
+
 /**
  * Spark's internal `PreciseTimestampConversion` reinterprets a value between the timestamp types
  * (`TimestampType` / `TimestampNTZType`) and `LongType` without losing microsecond precision. It

spark/src/main/scala/org/apache/comet/serde/literals.scala

@@ -24,7 +24,7 @@ import java.lang
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal}
 import org.apache.spark.sql.catalyst.util.ArrayData
-import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DateType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import com.google.protobuf.ByteString
@@ -56,7 +56,10 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
             .isInstanceOf[ArrayType])))) {
       Compatible(None)
     } else {
-      Unsupported(Some(s"Unsupported data type ${expr.dataType}"))
+      expr.dataType match {
+        case _: DayTimeIntervalType => Compatible(None)
+        case _ => Unsupported(Some(s"Unsupported data type ${expr.dataType}"))
+      }
     }
   }
 
@@ -78,7 +81,7 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
         case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte])
         case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short])
         case _: IntegerType | _: DateType => exprBuilder.setIntVal(value.asInstanceOf[Int])
-        case _: LongType | _: TimestampType | _: TimestampNTZType =>
+        case _: LongType | _: TimestampType | _: TimestampNTZType | _: DayTimeIntervalType =>
           exprBuilder.setLongVal(value.asInstanceOf[Long])
         case dt if isTimeType(dt) =>
           exprBuilder.setLongVal(value.asInstanceOf[Long])

spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala

@@ -219,7 +219,7 @@ class CometScalaUDFCodegen extends CometUDF with Logging {
       StructColumnSpec(nullable = true, fieldSpecs)
     case _: BitVector | _: TinyIntVector | _: SmallIntVector | _: IntVector | _: BigIntVector |
         _: Float4Vector | _: Float8Vector | _: DecimalVector | _: VarCharVector |
-        _: VarBinaryVector | _: DateDayVector | _: TimeStampMicroVector |
+        _: VarBinaryVector | _: DateDayVector | _: DurationVector | _: TimeStampMicroVector |
         _: TimeStampMicroTZVector =>
       ScalarColumnSpec(v.getClass.asInstanceOf[Class[_ <: ValueVector]], nullable = true)
     case other =>

spark/src/test/resources/sql-tests/expressions/datetime/multiply_dt_interval.sql

@@ -0,0 +1,67 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Routes multiply_dt_interval through the codegen dispatcher; produces DayTimeIntervalType.
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_multiply_dt_interval(days int, hours int, minutes int, seconds decimal(18,6), b tinyint, s smallint, i int, l long, f float, d double, dec decimal(10,2)) USING parquet
+
+statement
+INSERT INTO test_multiply_dt_interval VALUES
+  (1, 2, 3, 4.500000, CAST(2 AS TINYINT), CAST(3 AS SMALLINT), 2, CAST(3 AS BIGINT), CAST(1.5 AS FLOAT), CAST(2.5 AS DOUBLE), CAST(2.50 AS DECIMAL(10, 2))),
+  (-1, 0, 30, 15.250000, CAST(-2 AS TINYINT), CAST(-3 AS SMALLINT), -2, CAST(-3 AS BIGINT), CAST(-1.5 AS FLOAT), CAST(-2.5 AS DOUBLE), CAST(-2.50 AS DECIMAL(10, 2))),
+  (0, 0, 0, 0.000001, CAST(0 AS TINYINT), CAST(0 AS SMALLINT), 0, CAST(0 AS BIGINT), CAST(0.5 AS FLOAT), CAST(0.5 AS DOUBLE), CAST(0.50 AS DECIMAL(10, 2))),
+  (2, -6, 0, 0.000000, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
+
+query
+SELECT
+  make_dt_interval(days, hours, minutes, seconds) * b,
+  make_dt_interval(days, hours, minutes, seconds) * s,
+  make_dt_interval(days, hours, minutes, seconds) * i,
+  make_dt_interval(days, hours, minutes, seconds) * l,
+  make_dt_interval(days, hours, minutes, seconds) * f,
+  make_dt_interval(days, hours, minutes, seconds) * d,
+  make_dt_interval(days, hours, minutes, seconds) * dec
+FROM test_multiply_dt_interval
+
+-- literal interval input
+query
+SELECT INTERVAL '1 02:03:04.500000' DAY TO SECOND * i FROM test_multiply_dt_interval
+
+-- numeric on the left is normalized by Spark to multiply_dt_interval.
+query
+SELECT i * make_dt_interval(days, hours, minutes, seconds),
+  2 * INTERVAL '1 02:03:04.500000' DAY TO SECOND
+FROM test_multiply_dt_interval
+
+-- literal multipliers, including half-up rounding to the nearest microsecond.
+query
+SELECT
+  make_dt_interval(1, 2, 3, 4.5) * 2,
+  INTERVAL '0.000001' SECOND * 0.5D,
+  INTERVAL '0.000001' SECOND * CAST(0.50 AS DECIMAL(10, 2)),
+  make_dt_interval(-1, 0, 30, 15.25) * 1.5D
+
+-- null interval input
+query
+SELECT make_dt_interval(NULL, hours, minutes, seconds) * 2
+FROM test_multiply_dt_interval
+
+-- 106751991 days is valid, but multiplying it by two exceeds the int64 microsecond range.
+query expect_error(overflow)
+SELECT make_dt_interval(106751991) * 2