From 3c37026ed8ba85ff02b4716e61854c3919984343 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Wed, 24 Jun 2026 20:17:45 +0800
Subject: [PATCH 01/15] Support multiply_ym_interval with YearMonth interval
 codegen dispatch

---
 docs/source/user-guide/latest/datatypes.md    | 10 ++--
 docs/source/user-guide/latest/expressions.md  |  2 +-
 native/core/src/execution/planner.rs          | 20 +++++--
 native/core/src/execution/serde.rs            |  3 +-
 native/proto/src/proto/types.proto            |  1 +
 .../codegen/CometBatchKernelCodegen.scala     |  2 +
 .../CometBatchKernelCodegenInput.scala        | 18 ++++--
 .../CometBatchKernelCodegenOutput.scala       |  5 +-
 .../apache/comet/serde/QueryPlanSerde.scala   |  5 +-
 .../org/apache/comet/serde/datetime.scala     |  6 +-
 .../org/apache/comet/serde/literals.scala     |  7 ++-
 .../udf/codegen/CometScalaUDFCodegen.scala    |  8 +--
 .../apache/spark/sql/comet/util/Utils.scala   |  2 +
 .../datetime/multiply_ym_interval.sql         | 55 +++++++++++++++++++
 14 files changed, 116 insertions(+), 28 deletions(-)
 create mode 100644 spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 80465121daf..43d4042a8a2 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -83,11 +83,11 @@ the tables below and may be reconsidered based on demand:
 Interval types fall back to Spark today. Native acceleration is tracked by
 [#4540](https://github.com/apache/datafusion-comet/issues/4540).
 
-| Type                    | Status | Notes             |
-| ----------------------- | ------ | ----------------- |
-| `YearMonthIntervalType` | 🔜     | Tracked by #4540. |
-| `DayTimeIntervalType`   | 🔜     | Tracked by #4540. |
-| `CalendarIntervalType`  | 🔜     | Tracked by #4540. |
+| Type                    | Status | Notes                                                                  |
+| ----------------------- | ------ | ---------------------------------------------------------------------- |
+| `YearMonthIntervalType` | ✅     | Supported for `make_ym_interval` and YearMonth interval multiplication. |
+| `DayTimeIntervalType`   | 🔜     | Tracked by #4540.                                                      |
+| `CalendarIntervalType`  | 🔜     | Tracked by #4540.                                                      |
 
 ## Complex
 
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index b854e29d1f3..b32eda6d3b7 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -268,7 +268,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `make_timestamp` | ✅ |  |
 | `make_timestamp_ltz` | ✅ | 2-arg TIME form falls back |
 | `make_timestamp_ntz` | ✅ | 2-arg TIME form falls back |
-| `make_ym_interval` | 🔜 | [#4541](https://github.com/apache/datafusion-comet/issues/4541) |
+| `make_ym_interval` | ✅ | Routes through the JVM codegen dispatcher |
 | `minute` | ✅ |  |
 | `month` | ✅ |  |
 | `monthname` | ✅ | Abbreviated month name (Spark 4.0+) |
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index e89f0a8cf45..a1c2c108f6b 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -35,7 +35,9 @@ use crate::execution::{
 };
 use crate::jvm_bridge::{jni_call, JVMClasses};
 use arrow::compute::CastOptions;
-use arrow::datatypes::{DataType, Field, FieldRef, Schema, TimeUnit, DECIMAL128_MAX_PRECISION};
+use arrow::datatypes::{
+    DataType, Field, FieldRef, IntervalUnit, Schema, TimeUnit, DECIMAL128_MAX_PRECISION,
+};
 use arrow::ffi_stream::FFI_ArrowArrayStream;
 use datafusion::functions_aggregate::bit_and_or_xor::{bit_and_udaf, bit_or_udaf, bit_xor_udaf};
 use datafusion::functions_aggregate::count::count_udaf;
@@ -101,8 +103,8 @@ use datafusion::physical_expr::LexOrdering;
 use crate::parquet::parquet_exec::init_datasource_exec;
 use arrow::array::{
     new_empty_array, Array, ArrayRef, BinaryBuilder, BooleanArray, Date32Array, Decimal128Array,
-    Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, ListArray,
-    NullArray, StringBuilder, TimestampMicrosecondArray,
+    Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
+    IntervalYearMonthArray, ListArray, NullArray, StringBuilder, TimestampMicrosecondArray,
 };
 use arrow::buffer::{BooleanBuffer, NullBuffer, OffsetBuffer};
 use arrow::row::{OwnedRow, RowConverter, SortField};
@@ -362,6 +364,9 @@ impl PhysicalPlanner {
                         DataType::Time64(TimeUnit::Nanosecond) => {
                             ScalarValue::Time64Nanosecond(None)
                         }
+                        DataType::Interval(IntervalUnit::YearMonth) => {
+                            ScalarValue::IntervalYearMonth(None)
+                        }
                         dt => {
                             return Err(GeneralError(format!("{dt:?} is not supported in Comet")))
                         }
@@ -374,9 +379,12 @@ impl PhysicalPlanner {
                         Value::IntVal(value) => match data_type {
                             DataType::Int32 => ScalarValue::Int32(Some(*value)),
                             DataType::Date32 => ScalarValue::Date32(Some(*value)),
+                            DataType::Interval(IntervalUnit::YearMonth) => {
+                                ScalarValue::IntervalYearMonth(Some(*value))
+                            }
                             dt => {
                                 return Err(GeneralError(format!(
-                                    "Expected either 'Int32' or 'Date32' for IntVal, but found {dt:?}"
+                                    "Expected either 'Int32', 'Date32', or 'Interval(YearMonth)' for IntVal, but found {dt:?}"
                                 )))
                             }
                         },
@@ -3635,6 +3643,10 @@ fn literal_to_array_ref(
             list_literal.int_values.into(),
             Some(nulls.clone().into()),
         ))),
+        DataType::Interval(IntervalUnit::YearMonth) => Ok(Arc::new(IntervalYearMonthArray::new(
+            list_literal.int_values.into(),
+            Some(nulls.clone().into()),
+        ))),
         DataType::Timestamp(TimeUnit::Microsecond, None) => {
             Ok(Arc::new(TimestampMicrosecondArray::new(
                 list_literal.long_values.into(),
diff --git a/native/core/src/execution/serde.rs b/native/core/src/execution/serde.rs
index d6ec6be1323..86f7f2c1e7b 100644
--- a/native/core/src/execution/serde.rs
+++ b/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,7 @@ 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),
+        DataTypeId::YearMonthInterval => ArrowDataType::Interval(IntervalUnit::YearMonth),
         DataTypeId::Null => ArrowDataType::Null,
         DataTypeId::List => match dt_value
             .type_info
diff --git a/native/proto/src/proto/types.proto b/native/proto/src/proto/types.proto
index df0c0c5553d..2d9b45bb496 100644
--- a/native/proto/src/proto/types.proto
+++ b/native/proto/src/proto/types.proto
@@ -60,6 +60,7 @@ message DataType {
     MAP = 15;
     STRUCT = 16;
     TIME = 17;
+    YEAR_MONTH_INTERVAL = 18;
   }
   DataTypeId type_id = 1;
 
diff --git a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
index 1c990835bb7..ff4ff719937 100644
--- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
+++ b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
@@ -62,6 +62,7 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case "TinyIntVector" => classOf[TinyIntVector]
     case "SmallIntVector" => classOf[SmallIntVector]
     case "IntVector" => classOf[IntVector]
+    case "IntervalYearVector" => classOf[IntervalYearVector]
     case "BigIntVector" => classOf[BigIntVector]
     case "Float4Vector" => classOf[Float4Vector]
     case "Float8Vector" => classOf[Float8Vector]
@@ -82,6 +83,7 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case BooleanType | ByteType | ShortType | IntegerType | LongType => true
     case FloatType | DoubleType => true
     case _: DecimalType => true
+    case _: YearMonthIntervalType => true
     case _: StringType | _: BinaryType => true
     case DateType | TimestampType | TimestampNTZType => true
     case ArrayType(inner, _) => isSupportedDataType(inner)
diff --git a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala
index 09bfc52bd43..d9adf67d51a 100644
--- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala
+++ b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala
@@ -57,6 +57,7 @@ private[codegen] object CometBatchKernelCodegenInput {
     classOf[TinyIntVector],
     classOf[SmallIntVector],
     classOf[IntVector],
+    classOf[IntervalYearVector],
     classOf[BigIntVector],
     classOf[Float4Vector],
     classOf[Float8Vector],
@@ -127,7 +128,9 @@ private[codegen] object CometBatchKernelCodegenInput {
     }
     val intCases = withOrd.collect {
       case (ArrowColumnSpec(cls, _), ord)
-          if cls == classOf[IntVector] || cls == classOf[DateDayVector] =>
+          if cls == classOf[IntVector] ||
+            cls == classOf[DateDayVector] ||
+            cls == classOf[IntervalYearVector] =>
         s"      case $ord: return this.col$ord.getInt(this.rowIdx);"
     }
     val longCases = withOrd.collect {
@@ -590,7 +593,7 @@ private[codegen] object CometBatchKernelCodegenInput {
     case BooleanType => s"getBoolean($idx)"
     case ByteType => s"getByte($idx)"
     case ShortType => s"getShort($idx)"
-    case IntegerType | DateType => s"getInt($idx)"
+    case IntegerType | DateType | _: YearMonthIntervalType => s"getInt($idx)"
     case LongType | TimestampType | TimestampNTZType => s"getLong($idx)"
     case FloatType => s"getFloat($idx)"
     case DoubleType => s"getDouble($idx)"
@@ -687,7 +690,7 @@ private[codegen] object CometBatchKernelCodegenInput {
            |      public short getShort(int i) {
            |        return $childField.getShort(startIndex + i);
            |      }""".stripMargin
-      case IntegerType | DateType =>
+      case IntegerType | DateType | _: YearMonthIntervalType =>
         s"""      @Override
            |      public int getInt(int i) {
            |        return $childField.getInt(startIndex + i);
@@ -843,7 +846,7 @@ private[codegen] object CometBatchKernelCodegenInput {
           s"        case $fi: return ${path}_f$fi.getByte(this.rowIdx);"
         case ShortType =>
           s"        case $fi: return ${path}_f$fi.getShort(this.rowIdx);"
-        case IntegerType | DateType =>
+        case IntegerType | DateType | _: YearMonthIntervalType =>
           s"        case $fi: return ${path}_f$fi.getInt(this.rowIdx);"
         case LongType | TimestampType | TimestampNTZType =>
           s"        case $fi: return ${path}_f$fi.getLong(this.rowIdx);"
@@ -891,8 +894,11 @@ private[codegen] object CometBatchKernelCodegenInput {
           fieldReadScalar(fi, ShortType, f.nullable)
       }
     val intCases = scalarOrd.collect {
-      case (f, fi) if f.sparkType == IntegerType || f.sparkType == DateType =>
-        fieldReadScalar(fi, IntegerType, f.nullable)
+      case (f, fi)
+          if f.sparkType == IntegerType ||
+            f.sparkType == DateType ||
+            f.sparkType.isInstanceOf[YearMonthIntervalType] =>
+        fieldReadScalar(fi, f.sparkType, f.nullable)
     }
     val longCases = scalarOrd.collect {
       case (f, fi)
diff --git a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
index a26e3d17961..0da27f71665 100644
--- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
+++ b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
@@ -162,6 +162,7 @@ private[codegen] object CometBatchKernelCodegenOutput {
     case ByteType => classOf[TinyIntVector].getName
     case ShortType => classOf[SmallIntVector].getName
     case IntegerType => classOf[IntVector].getName
+    case _: YearMonthIntervalType => classOf[IntervalYearVector].getName
     case LongType => classOf[BigIntVector].getName
     case FloatType => classOf[Float4Vector].getName
     case DoubleType => classOf[Float8Vector].getName
@@ -208,7 +209,7 @@ 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 =>
       // Spark codegen emits the matching primitive Java type; Arrow `set` overloads accept it.
       val set = if (nested) "setSafe" else "set"
       OutputEmit("", s"$targetVec.$set($idx, $source);")
@@ -392,7 +393,7 @@ private[codegen] object CometBatchKernelCodegenOutput {
       case BooleanType => s"$target.getBoolean($idx)"
       case ByteType => s"$target.getByte($idx)"
       case ShortType => s"$target.getShort($idx)"
-      case IntegerType | DateType => s"$target.getInt($idx)"
+      case IntegerType | DateType | _: YearMonthIntervalType => s"$target.getInt($idx)"
       case LongType | TimestampType | TimestampNTZType => s"$target.getLong($idx)"
       case FloatType => s"$target.getFloat($idx)"
       case DoubleType => s"$target.getDouble($idx)"
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 143048fb44f..9e759a036ed 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -290,9 +290,11 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[Hour] -> CometHour,
       classOf[MakeDate] -> CometMakeDate,
       classOf[MakeTimestamp] -> CometMakeTimestamp,
+      classOf[MakeYMInterval] -> CometMakeYMInterval,
       classOf[MicrosToTimestamp] -> CometMicrosToTimestamp,
       classOf[MillisToTimestamp] -> CometMillisToTimestamp,
       classOf[MonthsBetween] -> CometMonthsBetween,
+      classOf[MultiplyYMInterval] -> CometMultiplyYMInterval,
       classOf[Minute] -> CometMinute,
       classOf[NextDay] -> CometNextDay,
       classOf[Second] -> CometSecond,
@@ -478,7 +480,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
-        _: DecimalType | _: DateType | _: BooleanType | _: NullType =>
+        _: DecimalType | _: DateType | _: BooleanType | _: NullType | _: YearMonthIntervalType =>
       true
     case dt if isTimeType(dt) =>
       true
@@ -517,6 +519,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       case _: MapType => 15
       case _: StructType => 16
       case dt if isTimeType(dt) => 17
+      case _: YearMonthIntervalType => 18
       case dt =>
         logWarning(s"Cannot serialize Spark data type: $dt")
         return None
diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
index bc6214a1882..2402ca529b9 100644
--- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala
+++ b/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, 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, FromUTCTimestamp, GetDateField, GetTimestamp, Hour, Hours, LastDay, Literal, MakeDate, MakeTimestamp, MakeYMInterval, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, MultiplyYMInterval, 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.{DataType, DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -852,6 +852,10 @@ object CometMonthsBetween extends CometCodegenDispatch[MonthsBetween]
 
 object CometMakeTimestamp extends CometCodegenDispatch[MakeTimestamp]
 
+object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval]
+
+object CometMultiplyYMInterval extends CometCodegenDispatch[MultiplyYMInterval]
+
 object CometMicrosToTimestamp extends CometCodegenDispatch[MicrosToTimestamp]
 
 object CometMillisToTimestamp extends CometCodegenDispatch[MillisToTimestamp]
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index 4f2a5dfa5e5..3c478b975fd 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/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, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType, YearMonthIntervalType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import com.google.protobuf.ByteString
@@ -77,7 +77,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
         case _: BooleanType => exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
         case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte])
         case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short])
-        case _: IntegerType | _: DateType => exprBuilder.setIntVal(value.asInstanceOf[Int])
+        case _: IntegerType | _: DateType | _: YearMonthIntervalType =>
+          exprBuilder.setIntVal(value.asInstanceOf[Int])
         case _: LongType | _: TimestampType | _: TimestampNTZType =>
           exprBuilder.setLongVal(value.asInstanceOf[Long])
         case dt if isTimeType(dt) =>
@@ -150,7 +151,7 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
             else null.asInstanceOf[Integer])
           listLiteralBuilder.addNullMask(casted != null)
         })
-      case IntegerType | DateType =>
+      case IntegerType | DateType | _: YearMonthIntervalType =>
         array.foreach(v => {
           val casted = v.asInstanceOf[Integer]
           listLiteralBuilder.addIntValues(casted)
diff --git a/spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala b/spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala
index f575dd5b53c..ad7d3037eee 100644
--- a/spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala
+++ b/spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala
@@ -217,10 +217,10 @@ class CometScalaUDFCodegen extends CometUDF with Logging {
           child = specFor(childVec))
       }
       StructColumnSpec(nullable = true, fieldSpecs)
-    case _: BitVector | _: TinyIntVector | _: SmallIntVector | _: IntVector | _: BigIntVector |
-        _: Float4Vector | _: Float8Vector | _: DecimalVector | _: VarCharVector |
-        _: VarBinaryVector | _: DateDayVector | _: TimeStampMicroVector |
-        _: TimeStampMicroTZVector =>
+    case _: BitVector | _: TinyIntVector | _: SmallIntVector | _: IntVector |
+        _: IntervalYearVector | _: BigIntVector | _: Float4Vector | _: Float8Vector |
+        _: DecimalVector | _: VarCharVector | _: VarBinaryVector | _: DateDayVector |
+        _: TimeStampMicroVector | _: TimeStampMicroTZVector =>
       ScalarColumnSpec(v.getClass.asInstanceOf[Class[_ <: ValueVector]], nullable = true)
     case other =>
       throw new UnsupportedOperationException(
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala b/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
index 15e1e2c410d..fdef21dad96 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
@@ -152,6 +152,8 @@ 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)
       case _ =>
         throw new UnsupportedOperationException(
           s"Unsupported data type: [${dt.getClass.getName}] ${dt.catalogString}")
diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql b/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql
new file mode 100644
index 00000000000..ef41ac31747
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql
@@ -0,0 +1,55 @@
+-- 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_ym_interval through the codegen dispatcher; produces YearMonthIntervalType.
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_multiply_ym_interval(y int, m int, i int, l long, f float, d double, dec decimal(10,2)) USING parquet
+
+statement
+INSERT INTO test_multiply_ym_interval VALUES
+  (1, 2, 2, CAST(3 AS BIGINT), CAST(1.5 AS FLOAT), CAST(2.5 AS DOUBLE), CAST(2.50 AS DECIMAL(10, 2))),
+  (-1, 1, -2, CAST(-3 AS BIGINT), CAST(-1.5 AS FLOAT), CAST(-2.5 AS DOUBLE), CAST(-2.50 AS DECIMAL(10, 2))),
+  (0, 6, 0, CAST(0 AS BIGINT), CAST(0.5 AS FLOAT), CAST(0.5 AS DOUBLE), CAST(0.50 AS DECIMAL(10, 2))),
+  (2, -6, NULL, NULL, NULL, NULL, NULL)
+
+query
+SELECT
+  make_ym_interval(y, m) * i,
+  make_ym_interval(y, m) * l,
+  make_ym_interval(y, m) * f,
+  make_ym_interval(y, m) * d,
+  make_ym_interval(y, m) * dec
+FROM test_multiply_ym_interval
+
+-- literal interval input
+query
+SELECT INTERVAL '1-2' YEAR TO MONTH * i FROM test_multiply_ym_interval
+
+-- numeric on the left is normalized by Spark to multiply_ym_interval.
+query
+SELECT i * make_ym_interval(y, m), 2 * INTERVAL '1-2' YEAR TO MONTH
+FROM test_multiply_ym_interval
+
+-- literal multipliers, including half-up rounding for fractional months.
+query
+SELECT
+  make_ym_interval(1, 2) * 2,
+  make_ym_interval(1, 2) * 1.5D,
+  make_ym_interval(1, 2) * CAST(1.50 AS DECIMAL(10, 2)),
+  make_ym_interval(-1, 1) * 1.5D

From b96f54749f81da4ed00dcb282c374bea4b926876 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Wed, 24 Jun 2026 20:28:29 +0800
Subject: [PATCH 02/15] Fix markdown formatting for interval docs

---
 docs/source/user-guide/latest/datatypes.md | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 43d4042a8a2..15d58d7f37e 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -80,14 +80,14 @@ the tables below and may be reconsidered based on demand:
 
 ## Interval
 
-Interval types fall back to Spark today. Native acceleration is tracked by
+Interval type support is incremental and tracked by
 [#4540](https://github.com/apache/datafusion-comet/issues/4540).
 
-| Type                    | Status | Notes                                                                  |
-| ----------------------- | ------ | ---------------------------------------------------------------------- |
+| Type                    | Status | Notes                                                                   |
+| ----------------------- | ------ | ----------------------------------------------------------------------- |
 | `YearMonthIntervalType` | ✅     | Supported for `make_ym_interval` and YearMonth interval multiplication. |
-| `DayTimeIntervalType`   | 🔜     | Tracked by #4540.                                                      |
-| `CalendarIntervalType`  | 🔜     | Tracked by #4540.                                                      |
+| `DayTimeIntervalType`   | 🔜     | Tracked by #4540.                                                       |
+| `CalendarIntervalType`  | 🔜     | Tracked by #4540.                                                       |
 
 ## Complex
 

From 01f04637d054a08af5f7ab1f0e81721f3e5c31e5 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Wed, 24 Jun 2026 20:28:29 +0800
Subject: [PATCH 03/15] Fix markdown formatting for interval docs

---
 docs/source/user-guide/latest/expressions.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index 30bd6d0f803..bb1be78b23a 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -397,7 +397,7 @@ expression-level). The `outer` variants are wired but marked `Incompatible`; the
 | Function | Status | Notes |
 | --- | --- | --- |
 | `%` | ✅ |  |
-| `*` | ✅ | Interval multiplication falls back |
+| `*` | ✅ | YearMonth interval multiplication routes through the JVM codegen dispatcher; DayTime/Calendar interval forms fall back |
 | `+` | ✅ |  |
 | `-` | ✅ |  |
 | `/` | ✅ |  |

From 1641d5709bcbcb86ace602ba5862f52bd173f339 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 12 Jul 2026 16:47:35 +0800
Subject: [PATCH 04/15] review by andy

---
 docs/source/user-guide/latest/datatypes.md           | 12 ++++++------
 .../comet/codegen/CometBatchKernelCodegen.scala      |  2 --
 .../codegen/CometBatchKernelCodegenOutput.scala      |  3 +--
 .../org/apache/comet/serde/QueryPlanSerde.scala      |  2 +-
 .../main/scala/org/apache/comet/serde/datetime.scala |  8 +++-----
 .../main/scala/org/apache/comet/serde/literals.scala |  2 +-
 .../expressions/datetime/multiply_ym_interval.sql    |  9 +++++++++
 7 files changed, 21 insertions(+), 17 deletions(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 15d58d7f37e..9ea4a7e768b 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -80,14 +80,14 @@ the tables below and may be reconsidered based on demand:
 
 ## Interval
 
-Interval type support is incremental and tracked by
+Interval types fall back to Spark today. Native acceleration is tracked by
 [#4540](https://github.com/apache/datafusion-comet/issues/4540).
 
-| Type                    | Status | Notes                                                                   |
-| ----------------------- | ------ | ----------------------------------------------------------------------- |
-| `YearMonthIntervalType` | ✅     | Supported for `make_ym_interval` and YearMonth interval multiplication. |
-| `DayTimeIntervalType`   | 🔜     | Tracked by #4540.                                                       |
-| `CalendarIntervalType`  | 🔜     | Tracked by #4540.                                                       |
+| Type                    | Status | Notes             |
+| ----------------------- | ------ | ----------------- |
+| `YearMonthIntervalType` | ✅      |                   |
+| `DayTimeIntervalType`   | ✅      |                   |
+| `CalendarIntervalType`  | 🔜     | Tracked by #4540. |
 
 ## Complex
 
diff --git a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
index 82251410493..107fb5e7f84 100644
--- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
+++ b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala
@@ -62,7 +62,6 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case "TinyIntVector" => classOf[TinyIntVector]
     case "SmallIntVector" => classOf[SmallIntVector]
     case "IntVector" => classOf[IntVector]
-    case "IntervalYearVector" => classOf[IntervalYearVector]
     case "BigIntVector" => classOf[BigIntVector]
     case "Float4Vector" => classOf[Float4Vector]
     case "Float8Vector" => classOf[Float8Vector]
@@ -85,7 +84,6 @@ object CometBatchKernelCodegen extends Logging with CometExprTraitShim {
     case BooleanType | ByteType | ShortType | IntegerType | LongType => true
     case FloatType | DoubleType => true
     case _: DecimalType => true
-    case _: YearMonthIntervalType => true
     case _: StringType | _: BinaryType => true
     case DateType | TimestampType | TimestampNTZType => true
     case _: YearMonthIntervalType | _: DayTimeIntervalType => true
diff --git a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
index ef926a319ca..fe282cdc656 100644
--- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
+++ b/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenOutput.scala
@@ -162,7 +162,6 @@ private[codegen] object CometBatchKernelCodegenOutput {
     case ByteType => classOf[TinyIntVector].getName
     case ShortType => classOf[SmallIntVector].getName
     case IntegerType => classOf[IntVector].getName
-    case _: YearMonthIntervalType => classOf[IntervalYearVector].getName
     case LongType => classOf[BigIntVector].getName
     case FloatType => classOf[Float4Vector].getName
     case DoubleType => classOf[Float8Vector].getName
@@ -397,7 +396,7 @@ private[codegen] object CometBatchKernelCodegenOutput {
       case BooleanType => s"$target.getBoolean($idx)"
       case ByteType => s"$target.getByte($idx)"
       case ShortType => s"$target.getShort($idx)"
-      case IntegerType | DateType | _: YearMonthIntervalType => s"$target.getInt($idx)"
+      case IntegerType | DateType => s"$target.getInt($idx)"
       case LongType | TimestampType | TimestampNTZType => s"$target.getLong($idx)"
       case FloatType => s"$target.getFloat($idx)"
       case DoubleType => s"$target.getDouble($idx)"
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index e27555754d6..185acc6de5b 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -505,7 +505,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
-        _: DecimalType | _: DateType | _: BooleanType | _: NullType | _: YearMonthIntervalType =>
+        _: DecimalType | _: DateType | _: BooleanType | _: NullType =>
       true
     case dt if isTimeType(dt) =>
       true
diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala b/spark/src/main/scala/org/apache/comet/serde/datetime.scala
index 84d48c84efe..3dc992a21ab 100644
--- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala
+++ b/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, MultiplyYMInterval, 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
@@ -920,10 +920,6 @@ object CometMakeTimestamp
   }
 }
 
-object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval]
-
-object CometMultiplyYMInterval extends CometCodegenDispatch[MultiplyYMInterval]
-
 object CometMicrosToTimestamp extends CometCodegenDispatch[MicrosToTimestamp]
 
 object CometMillisToTimestamp extends CometCodegenDispatch[MillisToTimestamp]
@@ -956,6 +952,8 @@ object CometGetTimestamp extends CometCodegenDispatch[GetTimestamp]
 
 object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval]
 
+object CometMultiplyYMInterval extends CometCodegenDispatch[MultiplyYMInterval]
+
 object CometMakeDTInterval extends CometCodegenDispatch[MakeDTInterval]
 
 /**
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index 3c478b975fd..f6ee3a0f061 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -42,7 +42,7 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
 
   override def getSupportLevel(expr: Literal): SupportLevel = {
 
-    if (supportedDataType(
+    if (expr.dataType.isInstanceOf[YearMonthIntervalType] || supportedDataType(
         expr.dataType,
         allowComplex = expr.value == null ||
 
diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql b/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql
index ef41ac31747..c64a1a786bb 100644
--- a/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql
+++ b/spark/src/test/resources/sql-tests/expressions/datetime/multiply_ym_interval.sql
@@ -53,3 +53,12 @@ SELECT
   make_ym_interval(1, 2) * 1.5D,
   make_ym_interval(1, 2) * CAST(1.50 AS DECIMAL(10, 2)),
   make_ym_interval(-1, 1) * 1.5D
+
+-- null interval input
+query
+SELECT make_ym_interval(NULL, m) * 2 FROM test_multiply_ym_interval
+
+-- 178956970 years and 7 months is Int.MaxValue months. Multiplication overflows regardless of
+-- ANSI mode; the lowercase pattern matches Spark 3.x and 4.x error messages.
+query expect_error(overflow)
+SELECT make_ym_interval(178956970, 7) * 2

From 49d882a2e4db143c2703a77db683f5499a80c83a Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 12 Jul 2026 17:01:01 +0800
Subject: [PATCH 05/15] fix md format

---
 docs/source/user-guide/latest/datatypes.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 9ea4a7e768b..0d90b80f29f 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -85,8 +85,8 @@ Interval types fall back to Spark today. Native acceleration is tracked by
 
 | Type                    | Status | Notes             |
 | ----------------------- | ------ | ----------------- |
-| `YearMonthIntervalType` | ✅      |                   |
-| `DayTimeIntervalType`   | ✅      |                   |
+| `YearMonthIntervalType` | ✅     |                   |
+| `DayTimeIntervalType`   | ✅     |                   |
 | `CalendarIntervalType`  | 🔜     | Tracked by #4540. |
 
 ## Complex

From 018efc1a7e357340654f67e84db09a91f07b09b4 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 12 Jul 2026 20:45:28 +0800
Subject: [PATCH 06/15] restructure the CometLiteral.getSupportLevel

---
 .../scala/org/apache/comet/serde/literals.scala   | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index f6ee3a0f061..8f9cbf04f31 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -41,22 +41,25 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
     "Not all data types are supported for literal values")
 
   override def getSupportLevel(expr: Literal): SupportLevel = {
-
-    if (expr.dataType.isInstanceOf[YearMonthIntervalType] || supportedDataType(
-        expr.dataType,
+    val dataType = expr.dataType
+    if (supportedDataType(
+        dataType,
         allowComplex = expr.value == null ||
 
           // Nested literal support for native reader
           // can be tracked https://github.com/apache/datafusion-comet/issues/1937
-          (expr.dataType
+          (dataType
             .isInstanceOf[ArrayType] && (!isComplexType(
-            expr.dataType.asInstanceOf[ArrayType].elementType) || expr.dataType
+            dataType.asInstanceOf[ArrayType].elementType) || dataType
             .asInstanceOf[ArrayType]
             .elementType
             .isInstanceOf[ArrayType])))) {
       Compatible(None)
     } else {
-      Unsupported(Some(s"Unsupported data type ${expr.dataType}"))
+      dataType match {
+        case _: YearMonthIntervalType => Compatible(None)
+        case _ => Unsupported(Some(s"Unsupported data type $dataType"))
+      }
     }
   }
 

From a9bc017b02946fafc6024a2a17038eaef085f860 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Mon, 20 Jul 2026 15:15:26 +0800
Subject: [PATCH 07/15] Update datatypes.md

---
 docs/source/user-guide/latest/datatypes.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 0d90b80f29f..8d118eee094 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -80,7 +80,7 @@ the tables below and may be reconsidered based on demand:
 
 ## Interval
 
-Interval types fall back to Spark today. Native acceleration is tracked by
+Calendar interval types fall back to Spark today. Native acceleration is tracked by
 [#4540](https://github.com/apache/datafusion-comet/issues/4540).
 
 | Type                    | Status | Notes             |

From 783e5eccecdc4d4a44364cbdc5b3d252ea809e54 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Mon, 20 Jul 2026 15:21:02 +0800
Subject: [PATCH 08/15] review suggestion

---
 spark/src/main/scala/org/apache/comet/serde/literals.scala | 1 +
 1 file changed, 1 insertion(+)

diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index 8f9cbf04f31..eda1a78e24b 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -57,6 +57,7 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
       Compatible(None)
     } else {
       dataType match {
+      // Keep YearMonthIntervalType out of QueryPlanSerde.supportedDataType, which gates broader native paths.
         case _: YearMonthIntervalType => Compatible(None)
         case _ => Unsupported(Some(s"Unsupported data type $dataType"))
       }

From 3c9867d1b8a11b797bd045e180717ded28cad53c Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Mon, 20 Jul 2026 15:40:18 +0800
Subject: [PATCH 09/15] fix style

---
 spark/src/main/scala/org/apache/comet/serde/literals.scala | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index eda1a78e24b..c860a1b519b 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -57,7 +57,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
       Compatible(None)
     } else {
       dataType match {
-      // Keep YearMonthIntervalType out of QueryPlanSerde.supportedDataType, which gates broader native paths.
+        // Keep YearMonthIntervalType out of QueryPlanSerde.supportedDataType, which gates broader
+        // native paths.
         case _: YearMonthIntervalType => Compatible(None)
         case _ => Unsupported(Some(s"Unsupported data type $dataType"))
       }

From 90e5ddd3e9c07e5d6e7aa13ebb5989b4a8bd0690 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 26 Jul 2026 15:35:01 +0800
Subject: [PATCH 10/15] minor change

---
 docs/source/user-guide/latest/datatypes.md                   | 2 +-
 .../main/scala/org/apache/comet/serde/QueryPlanSerde.scala   | 2 +-
 spark/src/main/scala/org/apache/comet/serde/literals.scala   | 5 +++--
 3 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 8d118eee094..11f0a3573f9 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -87,7 +87,7 @@ Calendar interval types fall back to Spark today. Native acceleration is tracked
 | ----------------------- | ------ | ----------------- |
 | `YearMonthIntervalType` | ✅     |                   |
 | `DayTimeIntervalType`   | ✅     |                   |
-| `CalendarIntervalType`  | 🔜     | Tracked by #4540. |
+| `CalendarIntervalType`  | ✅     |                   |
 
 ## Complex
 
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 0771a30ebab..1852fd4e06c 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -528,7 +528,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
-        _: DecimalType | _: DateType | _: BooleanType | _: NullType | CalendarIntervalType =>
+        _: DecimalType | _: DateType | _: BooleanType | _: NullType =>
       true
     case dt if isTimeType(dt) =>
       true
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index 47c3c14bedb..e40752dd5be 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/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, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType, YearMonthIntervalType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import com.google.protobuf.ByteString
@@ -59,7 +59,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
       dataType match {
         // Keep interval types out of QueryPlanSerde.supportedDataType, which gates broader native
         // paths.
-        case _: DayTimeIntervalType | _: YearMonthIntervalType => Compatible(None)
+        case _: DayTimeIntervalType | _: YearMonthIntervalType | CalendarIntervalType =>
+          Compatible(None)
         case _ => Unsupported(Some(s"Unsupported data type $dataType"))
       }
     }

From c540044310b43222ebdd33d787ed930718d51f39 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 26 Jul 2026 15:46:36 +0800
Subject: [PATCH 11/15] md prettier

---
 docs/source/user-guide/latest/datatypes.md | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md
index 11f0a3573f9..daaba637d4d 100644
--- a/docs/source/user-guide/latest/datatypes.md
+++ b/docs/source/user-guide/latest/datatypes.md
@@ -83,11 +83,11 @@ the tables below and may be reconsidered based on demand:
 Calendar interval types fall back to Spark today. Native acceleration is tracked by
 [#4540](https://github.com/apache/datafusion-comet/issues/4540).
 
-| Type                    | Status | Notes             |
-| ----------------------- | ------ | ----------------- |
-| `YearMonthIntervalType` | ✅     |                   |
-| `DayTimeIntervalType`   | ✅     |                   |
-| `CalendarIntervalType`  | ✅     |                   |
+| Type                    | Status | Notes |
+| ----------------------- | ------ | ----- |
+| `YearMonthIntervalType` | ✅     |       |
+| `DayTimeIntervalType`   | ✅     |       |
+| `CalendarIntervalType`  | ✅     |       |
 
 ## Complex
 

From b6013da7d8dd3a4481bb806af9af6958f9039025 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 26 Jul 2026 20:16:48 +0800
Subject: [PATCH 12/15] Use DataTypeSupport for sink type checks

---
 .../org/apache/comet/serde/operator/CometSink.scala    | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
index ed69f82e2ab..140b241dd6e 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
@@ -19,6 +19,7 @@
 
 package org.apache.comet.serde.operator
 
+import scala.collection.mutable.ListBuffer
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.comet.{CometNativeExec, CometSinkPlaceHolder}
@@ -28,7 +29,7 @@ import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec
 import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
 import org.apache.spark.sql.types.DataType
 
-import org.apache.comet.CometConf
+import org.apache.comet.{CometConf, DataTypeSupport}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.ConfigEntry
 import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
@@ -39,7 +40,7 @@ import org.apache.comet.serde.QueryPlanSerde.{serializeDataType, supportedDataTy
  * CometSink is the base class for transformations from a Spark operator to a Comet operator where
  * the native plan is a ScanExec that will read data from the Comet operator running the JVM.
  */
-abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
+abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] with DataTypeSupport {
 
   override def enabledConfig: Option[ConfigEntry[Boolean]] = None
 
@@ -54,8 +55,9 @@ abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
       op: T,
       builder: Operator.Builder,
       childOp: OperatorOuterClass.Operator*): Option[OperatorOuterClass.Operator] = {
-    val supportedTypes =
-      op.output.forall(a => supportedDataType(a.dataType, allowComplex = true))
+    val supportedTypes = op.output.forall(a =>
+      supportedDataType(a.dataType, allowComplex = true) ||
+        isTypeSupported(a.dataType, a.name, ListBuffer.empty))
 
     if (!supportedTypes) {
       withFallbackReason(op, "Unsupported data type")

From a2349c590136fd87788189c6e8a25bda4947e7f6 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Fri, 28 Aug 2026 09:41:42 +0800
Subject: [PATCH 13/15] address review: keep CalendarIntervalType support
 intact, group interval expressions

Restore CalendarIntervalType in QueryPlanSerde.supportedDataType (added
deliberately by #4898; removing it narrowed hash, scalar-subquery, and
nested-type gates out of scope for this PR) and revert CometSink to main,
since its isTypeSupported OR existed only to patch sinks around that
removal. CometLiteral now only adds YearMonthIntervalType alongside the
existing DayTimeIntervalType arm. Move MultiplyYMInterval next to
MultiplyDTInterval in the expression map's interval cluster.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../scala/org/apache/comet/serde/QueryPlanSerde.scala    | 4 ++--
 .../src/main/scala/org/apache/comet/serde/literals.scala | 9 +++++----
 .../org/apache/comet/serde/operator/CometSink.scala      | 7 +++----
 3 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 7518b675568..72f7e4f0ac5 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -305,12 +305,12 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[MakeDTInterval] -> CometMakeDTInterval,
       classOf[MakeInterval] -> CometMakeInterval,
       classOf[MultiplyDTInterval] -> CometMultiplyDTInterval,
+      classOf[MultiplyYMInterval] -> CometMultiplyYMInterval,
       classOf[TimestampAdd] -> CometTimestampAdd,
       classOf[TimestampDiff] -> CometTimestampDiff,
       classOf[MicrosToTimestamp] -> CometMicrosToTimestamp,
       classOf[MillisToTimestamp] -> CometMillisToTimestamp,
       classOf[MonthsBetween] -> CometMonthsBetween,
-      classOf[MultiplyYMInterval] -> CometMultiplyYMInterval,
       classOf[Minute] -> CometMinute,
       classOf[NextDay] -> CometNextDay,
       classOf[PreciseTimestampConversion] -> CometPreciseTimestampConversion,
@@ -535,7 +535,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: TimestampNTZType |
-        _: DecimalType | _: DateType | _: BooleanType | _: NullType =>
+        _: DecimalType | _: DateType | _: BooleanType | _: NullType | CalendarIntervalType =>
       true
     case dt if isTimeType(dt) =>
       true
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index e40752dd5be..08ee0b3c46a 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/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, CalendarIntervalType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DateType, DayTimeIntervalType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampNTZType, TimestampType, YearMonthIntervalType}
 import org.apache.spark.unsafe.types.UTF8String
 
 import com.google.protobuf.ByteString
@@ -57,9 +57,10 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
       Compatible(None)
     } else {
       dataType match {
-        // Keep interval types out of QueryPlanSerde.supportedDataType, which gates broader native
-        // paths.
-        case _: DayTimeIntervalType | _: YearMonthIntervalType | CalendarIntervalType =>
+        // ANSI interval types are deliberately kept out of QueryPlanSerde.supportedDataType so
+        // they are not claimed as flowing through arbitrary native operators; literals are
+        // supported here.
+        case _: DayTimeIntervalType | _: YearMonthIntervalType =>
           Compatible(None)
         case _ => Unsupported(Some(s"Unsupported data type $dataType"))
       }
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
index e6bfd83539b..f2026013cbf 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala
@@ -19,7 +19,6 @@
 
 package org.apache.comet.serde.operator
 
-import scala.collection.mutable.ListBuffer
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.comet.{CometNativeExec, CometSinkPlaceHolder}
@@ -29,7 +28,7 @@ import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec
 import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
 import org.apache.spark.sql.types.{ArrayType, DataType, DayTimeIntervalType, MapType, StructType, YearMonthIntervalType}
 
-import org.apache.comet.{CometConf, DataTypeSupport}
+import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.ConfigEntry
 import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
@@ -40,7 +39,7 @@ import org.apache.comet.serde.QueryPlanSerde.{serializeDataType, supportedDataTy
  * CometSink is the base class for transformations from a Spark operator to a Comet operator where
  * the native plan is a ScanExec that will read data from the Comet operator running the JVM.
  */
-abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] with DataTypeSupport {
+abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
 
   override def enabledConfig: Option[ConfigEntry[Boolean]] = None
 
@@ -51,7 +50,7 @@ abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] with Data
     case ArrayType(elementType, _) => supportedSinkDataType(elementType)
     case MapType(keyType, valueType, _) =>
       supportedSinkDataType(keyType) && supportedSinkDataType(valueType)
-    case _ => supportedDataType(dt) || isTypeSupported(dt, "", ListBuffer.empty)
+    case _ => supportedDataType(dt)
   }
 
   /**

From b633a6f56643d89ac7f19e37c1ae6117e6772e8e Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 13 Sep 2026 00:37:13 +0800
Subject: [PATCH 14/15] fix: admit year-month intervals in the list-literal
 gate

Merging upstream/main brought in `listLiteralElementSupported`, which
`CometLiteralSuite` pins against `makeListLiteral`'s arms in both directions.
This branch had already added a `YearMonthIntervalType` arm to the encoder, so
after the merge the two sides disagreed and the pin failed.

Widen the gate rather than drop the encoder arm: `serializeDataType` already
maps the type and recurses through `ArrayType`, and `literal_to_array_ref`
already reads the ints back as an `IntervalYearMonthArray`. An
`array<interval year to month>` literal is now serialized directly instead of
sending the whole projection back to Spark with "Unsupported data type
ArrayType(YearMonthIntervalType(0,1),true)".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../scala/org/apache/comet/serde/literals.scala     |  3 +++
 .../org/apache/comet/serde/CometLiteralSuite.scala  | 13 +++++++++++++
 2 files changed, 16 insertions(+)

diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
index 0bf15db656a..a46c431a48c 100644
--- a/spark/src/main/scala/org/apache/comet/serde/literals.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -245,6 +245,9 @@ object CometLiteral extends CometExpressionSerde[Literal] with CometTypeShim wit
         TimestampType | TimestampNTZType | FloatType | DoubleType | StringType | BinaryType =>
       true
     case _: DecimalType => true
+    // Matched as a type rather than a stable identifier: the start/end fields participate in
+    // `equals`, and every (start, end) pair is carried as the same month count.
+    case _: YearMonthIntervalType => true
     case ArrayType(elementType, _) => listLiteralElementSupported(elementType)
     case _ => false
   }
diff --git a/spark/src/test/scala/org/apache/comet/serde/CometLiteralSuite.scala b/spark/src/test/scala/org/apache/comet/serde/CometLiteralSuite.scala
index 746cc8e4b88..c58e65067f5 100644
--- a/spark/src/test/scala/org/apache/comet/serde/CometLiteralSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/serde/CometLiteralSuite.scala
@@ -123,4 +123,17 @@ class CometLiteralSuite extends CometTestBase with CometTypeShim {
     assert(!CometLiteral.listLiteralElementSupported(collated))
     assert(!encoderAcceptsElement(Array.empty, ArrayType(collated)))
   }
+
+  // A year-month interval rides in `int_values` as the month count `IntegerType` uses, and
+  // `literal_to_array_ref` reads those ints back as an `IntervalYearMonthArray`, so the whole
+  // array literal is serialized directly instead of sending the projection back to Spark.
+  // `needsExpansion` has no arm for it either, so declining it here is a fallback, not a rewrite.
+  test("a year-month interval array literal is serialized rather than declined") {
+    withParquetTable(Seq((1, 2), (3, 4)), "tbl") {
+      checkSparkAnswerAndOperator(
+        sql("SELECT array(INTERVAL '1-2' YEAR TO MONTH, INTERVAL '-3' MONTH, NULL) FROM tbl"))
+      checkSparkAnswerAndOperator(
+        sql("SELECT array(array(INTERVAL '2-1' YEAR TO MONTH), array()) FROM tbl"))
+    }
+  }
 }

From 68c299d991a85c828534c9bc08a9d267c3d86028 Mon Sep 17 00:00:00 2001
From: peterxcli <peterxcli@gmail.com>
Date: Sun, 13 Sep 2026 13:39:20 +0800
Subject: [PATCH 15/15] test: expect native execution for a null year-month
 interval in abs.sql

`abs.sql` pinned `SELECT abs(CAST(NULL AS INTERVAL YEAR TO MONTH))` as a
fallback because CometLiteral did not admit YearMonthIntervalType literals,
noting that it "flips to a failure when it is fixed". This branch adds that
literal support, so the query now runs natively and the pinned fallback
failed in every [expressions] CI job (Spark 3.4, 3.5, 4.0, 4.1, 4.2). Assert
the answer and native operators instead.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/test/resources/sql-tests/expressions/math/abs.sql  | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs.sql b/spark/src/test/resources/sql-tests/expressions/math/abs.sql
index ff28bfd3e9e..c32cadd69ed 100644
--- a/spark/src/test/resources/sql-tests/expressions/math/abs.sql
+++ b/spark/src/test/resources/sql-tests/expressions/math/abs.sql
@@ -56,10 +56,9 @@ INSERT INTO test_abs_iv_overflow VALUES (1, 2, 3, 4.5), (-106751991, -4, 0, -54.
 query expect_error(overflow)
 SELECT abs(make_dt_interval(d, h, m, s)) FROM test_abs_iv_overflow
 
--- pinned fallback: NullPropagation folds the ym null into a bare typed literal that CometLiteral
--- does not admit (it special-cases only DayTimeIntervalType, literals.scala:63), so the whole
--- projection falls back. Pre-existing gap (#5061); this flips to a failure when it is fixed.
-query expect_fallback(Unsupported data type YearMonthIntervalType)
+-- NullPropagation folds the ym null into a bare typed literal. CometLiteral admits
+-- YearMonthIntervalType literals (one of the gaps tracked in #5061), so it stays native.
+query
 SELECT abs(CAST(NULL AS INTERVAL YEAR TO MONTH))
 
 query