feat: support interval types and make_ym_interval / make_dt_interval

native/core/src/execution/serde.rs

@@ -19,7 +19,7 @@
 
 use super::operators::ExecutionError;
 use crate::errors::ExpressionError;
-use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
+use arrow::datatypes::{DataType as ArrowDataType, IntervalUnit, TimeUnit};
 use arrow::datatypes::{Field, Fields};
 use datafusion_comet_proto::{
     spark_config, spark_expression,
@@ -97,6 +97,11 @@ pub fn to_arrow_datatype(dt_value: &DataType) -> ArrowDataType {
         DataTypeId::TimestampNtz => ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
         DataTypeId::Date => ArrowDataType::Date32,
         DataTypeId::Time => ArrowDataType::Time64(TimeUnit::Nanosecond),
+        // Spark's YearMonthIntervalType maps to Arrow Interval(YearMonth) (int32 months).
+        DataTypeId::YearMonthInterval => ArrowDataType::Interval(IntervalUnit::YearMonth),
+        // Spark's DayTimeIntervalType stores microseconds in an int64, which matches Arrow
+        // Duration(Microsecond) rather than the lossy Interval(DayTime) {days, millis} layout.
+        DataTypeId::DayTimeInterval => ArrowDataType::Duration(TimeUnit::Microsecond),
         DataTypeId::Null => ArrowDataType::Null,
         DataTypeId::List => match dt_value
             .type_info

native/proto/src/proto/types.proto

@@ -60,6 +60,8 @@ message DataType {
     MAP = 15;
     STRUCT = 16;
     TIME = 17;
+    YEAR_MONTH_INTERVAL = 18;
+    DAY_TIME_INTERVAL = 19;
   }
   DataTypeId type_id = 1;
 

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

@@ -71,6 +71,8 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case "TimeStampMicroTZVector" => classOf[TimeStampMicroTZVector]
     case "VarCharVector" => classOf[VarCharVector]
     case "VarBinaryVector" => classOf[VarBinaryVector]
+    case "IntervalYearVector" => classOf[IntervalYearVector]
+    case "DurationVector" => classOf[DurationVector]
     case other => throw new IllegalArgumentException(s"unknown Arrow vector class: $other")
   }
 
@@ -84,6 +86,7 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case _: DecimalType => true
     case _: StringType | _: BinaryType => true
     case DateType | TimestampType | TimestampNTZType => true
+    case _: YearMonthIntervalType | _: DayTimeIntervalType => true
     case ArrayType(inner, _) => isSupportedDataType(inner)
     case st: StructType => st.fields.forall(f => isSupportedDataType(f.dataType))
     case mt: MapType => isSupportedDataType(mt.keyType) && isSupportedDataType(mt.valueType)

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

@@ -171,6 +171,8 @@ private[codegen] object CometBatchKernelCodegenOutput {
     case DateType => classOf[DateDayVector].getName
     case TimestampType => classOf[TimeStampMicroTZVector].getName
     case TimestampNTZType => classOf[TimeStampMicroVector].getName
+    case _: YearMonthIntervalType => classOf[IntervalYearVector].getName
+    case _: DayTimeIntervalType => classOf[DurationVector].getName
     case _: ArrayType => classOf[ListVector].getName
     case _: StructType => classOf[StructVector].getName
     case _: MapType => classOf[MapVector].getName
@@ -208,8 +210,10 @@ private[codegen] object CometBatchKernelCodegenOutput {
       val set = if (nested) "setSafe" else "set"
       OutputEmit("", s"$targetVec.$set($idx, $source ? 1 : 0);")
     case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | DateType |
-        TimestampType | TimestampNTZType =>
+        TimestampType | TimestampNTZType | _: YearMonthIntervalType | _: DayTimeIntervalType =>
       // Spark codegen emits the matching primitive Java type; Arrow `set` overloads accept it.
+      // YearMonthIntervalType -> IntervalYearVector.set(int, int months);
+      // DayTimeIntervalType -> DurationVector.set(int, long micros).
       val set = if (nested) "setSafe" else "set"
       OutputEmit("", s"$targetVec.$set($idx, $source);")
     case dt: DecimalType =>

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

@@ -291,6 +291,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[Hour] -> CometHour,
       classOf[MakeDate] -> CometMakeDate,
       classOf[MakeTimestamp] -> CometMakeTimestamp,
+      classOf[MakeYMInterval] -> CometMakeYMInterval,
+      classOf[MakeDTInterval] -> CometMakeDTInterval,
       classOf[MicrosToTimestamp] -> CometMicrosToTimestamp,
       classOf[MillisToTimestamp] -> CometMillisToTimestamp,
       classOf[MonthsBetween] -> CometMonthsBetween,
@@ -519,6 +521,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       case _: MapType => 15
       case _: StructType => 16
       case dt if isTimeType(dt) => 17
+      case _: YearMonthIntervalType => 18
+      case _: DayTimeIntervalType => 19
       case dt =>
         logWarning(s"Cannot serialize Spark data type: $dt")
         return None

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, MakeTimestamp, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, NextDay, 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, NextDay, 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.{DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -949,3 +949,7 @@ object CometToUnixTimestamp
 }
 
 object CometGetTimestamp extends CometCodegenDispatch[GetTimestamp]
+
+object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval]
+
+object CometMakeDTInterval extends CometCodegenDispatch[MakeDTInterval]

spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala

@@ -109,6 +109,7 @@ object Utils extends CometTypeShim with Logging {
     case yi: ArrowType.Interval if yi.getUnit == IntervalUnit.YEAR_MONTH =>
       YearMonthIntervalType()
     case di: ArrowType.Interval if di.getUnit == IntervalUnit.DAY_TIME => DayTimeIntervalType()
+    case d: ArrowType.Duration if d.getUnit == TimeUnit.MICROSECOND => DayTimeIntervalType()
     case t: ArrowType.Time if t.getUnit == TimeUnit.NANOSECOND && t.getBitWidth == 64 =>
       // scalastyle:off classforname
       val clazz = Class.forName("org.apache.spark.sql.types.TimeType$")
@@ -152,6 +153,10 @@ object Utils extends CometTypeShim with Logging {
       case NullType => ArrowType.Null.INSTANCE
       case dt if isTimeType(dt) =>
         new ArrowType.Time(TimeUnit.NANOSECOND, 64)
+      case _: YearMonthIntervalType => new ArrowType.Interval(IntervalUnit.YEAR_MONTH)
+      // Spark stores DayTimeIntervalType as microseconds in an int64, matching Arrow
+      // Duration(Microsecond) rather than the lossy Interval(DayTime) {days, millis} layout.
+      case _: DayTimeIntervalType => new ArrowType.Duration(TimeUnit.MICROSECOND)
       case _ =>
         throw new UnsupportedOperationException(
           s"Unsupported data type: [${dt.getClass.getName}] ${dt.catalogString}")

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

@@ -0,0 +1,43 @@
+-- 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 make_dt_interval through the codegen dispatcher; produces DayTimeIntervalType.
+
+statement
+CREATE TABLE test_mdi(d int, h int, mi int, s decimal(18,6)) USING parquet
+
+statement
+INSERT INTO test_mdi VALUES (1, 2, 3, 4.5), (0, 0, 0, 0), (-1, 0, 30, 15.250), (NULL, 1, 1, 1)
+
+query
+SELECT d, h, mi, s, make_dt_interval(d, h, mi, s) FROM test_mdi
+
+-- literal arguments
+query
+SELECT make_dt_interval(1, 2, 3, 4.5), make_dt_interval(0, 0, 0, 0)
+
+-- default arguments
+query
+SELECT make_dt_interval(1), make_dt_interval(1, 2), make_dt_interval()
+
+-- overflow: days * MICROS_PER_DAY exceeds the int64 microsecond range. makeDayTimeInterval throws
+-- unconditionally (not ANSI-gated); this confirms the dispatched codegen path propagates Spark's
+-- exception. The pattern is the lowercase word so it matches every version: Spark 4.x raises
+-- INTERVAL_ARITHMETIC_OVERFLOW ("Integer overflow while operating with intervals") while Spark 3.x
+-- raises a raw ArithmeticException ("long overflow").
+query expect_error(overflow)
+SELECT make_dt_interval(2147483647)

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

@@ -0,0 +1,43 @@
+-- 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 make_ym_interval through the codegen dispatcher; produces YearMonthIntervalType.
+
+statement
+CREATE TABLE test_myi(y int, m int) USING parquet
+
+statement
+INSERT INTO test_myi VALUES (1, 2), (0, 0), (-1, 6), (5, -3), (NULL, 3), (2, NULL)
+
+query
+SELECT y, m, make_ym_interval(y, m) FROM test_myi
+
+-- literal arguments
+query
+SELECT make_ym_interval(1, 2), make_ym_interval(0, 0), make_ym_interval(-5, 11)
+
+-- default arguments
+query
+SELECT make_ym_interval(3), make_ym_interval()
+
+-- overflow: years * 12 exceeds Int range. makeYearMonthInterval throws unconditionally (not
+-- ANSI-gated); this confirms the dispatched codegen path propagates Spark's exception. The
+-- pattern is the lowercase word so it matches every version: Spark 4.x raises
+-- INTERVAL_ARITHMETIC_OVERFLOW ("Integer overflow while operating with intervals") while Spark 3.x
+-- raises a raw ArithmeticException ("integer overflow").
+query expect_error(overflow)
+SELECT make_ym_interval(2147483647)