feat: full native make_interval with 2.6x faster than spark and 1.1x faster than codegen dispatch

docs/source/user-guide/latest/expressions.md

@@ -277,7 +277,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `localtimestamp` | ✅ | — |  |
 | `make_date` | ✅ | Native |  |
 | `make_dt_interval` | ✅ | Codegen dispatch |  |
-| `make_interval` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; intervals outside Arrow's nanosecond range are tracked by [#5279](https://github.com/apache/datafusion-comet/issues/5279); the native path is opt-in via allowIncompatible ([details](compatibility/expressions/datetime.md)) |
+| `make_interval` | ✅ | Native |  |
 | `make_time` | 🔜 | — | Spark 4.1 TIME type; tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) |
 | `make_timestamp` | ✅ | Hybrid |  |
 | `make_timestamp_ltz` | ✅ | — | 2-arg TIME form falls back |

native/core/src/execution/serde.rs

@@ -31,6 +31,7 @@ use datafusion_comet_proto::{
     spark_expression::DataType,
     spark_operator,
 };
+use datafusion_comet_spark_expr::calendar_interval_type;
 use prost::Message;
 use std::{io::Cursor, sync::Arc};
 
@@ -102,9 +103,7 @@ pub fn to_arrow_datatype(dt_value: &DataType) -> ArrowDataType {
         // 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),
-        // Spark's CalendarIntervalType stores months, days, and microseconds. Arrow stores the
-        // same components with nanosecond precision.
-        DataTypeId::CalendarInterval => ArrowDataType::Interval(IntervalUnit::MonthDayNano),
+        DataTypeId::CalendarInterval => calendar_interval_type(),
         DataTypeId::Null => ArrowDataType::Null,
         DataTypeId::List => match dt_value
             .type_info

native/spark-expr/Cargo.toml

@@ -30,7 +30,6 @@ edition = { workspace = true }
 arrow = { workspace = true }
 chrono = { workspace = true }
 datafusion = { workspace = true }
-datafusion-spark = { workspace = true }
 chrono-tz = { workspace = true }
 num = { workspace = true }
 regex = { workspace = true }
@@ -222,4 +221,4 @@ harness = false
 
 [[bench]]
 name = "cast_int_to_decimal"
-harness = false
+harness = false

native/spark-expr/src/datetime_funcs/make_interval.rs

@@ -16,72 +16,225 @@
 // under the License.
 
 use crate::arithmetic_overflow_error;
-use arrow::array::Array;
-use arrow::datatypes::DataType;
-use datafusion::common::Result;
-use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature};
-use datafusion_spark::function::datetime::make_interval::SparkMakeInterval as DataFusionMakeInterval;
+use arrow::array::{Array, ArrayRef, Decimal128Array, Int32Array, Int64Array, StructArray};
+use arrow::buffer::NullBuffer;
+use arrow::datatypes::{DataType, Field, Fields};
+use datafusion::common::{DataFusionError, Result};
+use datafusion::logical_expr::{
+    ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
+};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+const CALENDAR_INTERVAL_STRUCT_KEY: &str = "SPARK::calendarInterval::struct";
+const MICROS_PER_HOUR: i64 = 3_600_000_000;
+const MICROS_PER_MINUTE: i64 = 60_000_000;
+
+pub fn calendar_interval_type() -> DataType {
+    let months = Field::new("months", DataType::Int32, false).with_metadata(HashMap::from([(
+        CALENDAR_INTERVAL_STRUCT_KEY.to_string(),
+        "true".to_string(),
+    )]));
+    DataType::Struct(Fields::from(vec![
+        months,
+        Field::new("days", DataType::Int32, false),
+        Field::new("microseconds", DataType::Int64, false),
+    ]))
+}
 
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct SparkMakeInterval {
-    inner: DataFusionMakeInterval,
+    signature: Signature,
     fail_on_error: bool,
 }
 
 impl SparkMakeInterval {
     pub fn new(fail_on_error: bool) -> Self {
         Self {
-            inner: DataFusionMakeInterval::new(),
+            signature: Signature::exact(
+                vec![
+                    DataType::Int32,
+                    DataType::Int32,
+                    DataType::Int32,
+                    DataType::Int32,
+                    DataType::Int32,
+                    DataType::Int32,
+                    DataType::Decimal128(18, 6),
+                ],
+                Volatility::Immutable,
+            ),
             fail_on_error,
         }
     }
 }
 
+fn make_interval(
+    years: i32,
+    months: i32,
+    weeks: i32,
+    days: i32,
+    hours: i32,
+    minutes: i32,
+    seconds_micros: i128,
+) -> std::result::Result<(i32, i32, i64), &'static str> {
+    let months = years
+        .checked_mul(12)
+        .and_then(|years| years.checked_add(months))
+        .ok_or("integer")?;
+    let days = weeks
+        .checked_mul(7)
+        .and_then(|weeks| weeks.checked_add(days))
+        .ok_or("integer")?;
+    let micros = i64::try_from(seconds_micros)
+        .ok()
+        .and_then(|seconds| {
+            i64::from(hours)
+                .checked_mul(MICROS_PER_HOUR)
+                .and_then(|hours| seconds.checked_add(hours))
+        })
+        .and_then(|micros| {
+            i64::from(minutes)
+                .checked_mul(MICROS_PER_MINUTE)
+                .and_then(|minutes| micros.checked_add(minutes))
+        })
+        .ok_or("long")?;
+    Ok((months, days, micros))
+}
+
 impl ScalarUDFImpl for SparkMakeInterval {
     fn name(&self) -> &str {
-        self.inner.name()
+        "make_interval"
     }
 
     fn signature(&self) -> &Signature {
-        self.inner.signature()
+        &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        self.inner.return_type(arg_types)
+    fn return_type(&self, _: &[DataType]) -> Result<DataType> {
+        Ok(calendar_interval_type())
     }
 
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
-        let inputs = if self.fail_on_error {
-            Some(args.args.clone())
-        } else {
-            None
-        };
-        let result = self.inner.invoke_with_args(args)?;
-
-        if let Some(inputs) = inputs {
-            let inputs_are_valid = |i| {
-                inputs.iter().all(|input| match input {
-                    ColumnarValue::Array(values) => values.is_valid(i),
-                    ColumnarValue::Scalar(value) => !value.is_null(),
-                })
-            };
-            let overflow = match &result {
-                ColumnarValue::Array(values) => values.nulls().is_some_and(|nulls| {
-                    nulls.null_count() != 0
-                        && nulls
-                            .iter()
-                            .enumerate()
-                            .any(|(i, is_valid)| !is_valid && inputs_are_valid(i))
-                }),
-                ColumnarValue::Scalar(value) => value.is_null() && inputs_are_valid(0),
-            };
-            if overflow {
-                // Spark identifies the integer or long operation that overflowed. The native
-                // wrapper only sees the result null mask, so it can only report interval overflow.
-                return Err(arithmetic_overflow_error("interval").into());
+        let number_rows = args.number_rows;
+        let arrays = args
+            .args
+            .into_iter()
+            .map(|arg| arg.into_array(number_rows))
+            .collect::<Result<Vec<_>>>()?;
+        let years = arrays[0]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 years".into())
+            })?;
+        let months = arrays[1]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 months".into())
+            })?;
+        let weeks = arrays[2]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 weeks".into())
+            })?;
+        let days = arrays[3]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 days".into())
+            })?;
+        let hours = arrays[4]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 hours".into())
+            })?;
+        let minutes = arrays[5]
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Int32 minutes".into())
+            })?;
+        let seconds = arrays[6]
+            .as_any()
+            .downcast_ref::<Decimal128Array>()
+            .ok_or_else(|| {
+                DataFusionError::Execution("make_interval: expected Decimal128 seconds".into())
+            })?;
+
+        let mut result_months = Vec::with_capacity(years.len());
+        let mut result_days = Vec::with_capacity(years.len());
+        let mut result_micros = Vec::with_capacity(years.len());
+        let mut valid = Vec::with_capacity(years.len());
+
+        for i in 0..years.len() {
+            if arrays.iter().any(|array| array.is_null(i)) {
+                result_months.push(0);
+                result_days.push(0);
+                result_micros.push(0);
+                valid.push(false);
+                continue;
+            }
+
+            match make_interval(
+                years.value(i),
+                months.value(i),
+                weeks.value(i),
+                days.value(i),
+                hours.value(i),
+                minutes.value(i),
+                seconds.value(i),
+            ) {
+                Ok((months, days, micros)) => {
+                    result_months.push(months);
+                    result_days.push(days);
+                    result_micros.push(micros);
+                    valid.push(true);
+                }
+                Err(from_type) if self.fail_on_error => {
+                    return Err(arithmetic_overflow_error(from_type).into());
+                }
+                Err(_) => {
+                    result_months.push(0);
+                    result_days.push(0);
+                    result_micros.push(0);
+                    valid.push(false);
+                }
             }
         }
 
-        Ok(result)
+        let columns: Vec<ArrayRef> = vec![
+            Arc::new(Int32Array::from(result_months)),
+            Arc::new(Int32Array::from(result_days)),
+            Arc::new(Int64Array::from(result_micros)),
+        ];
+        let DataType::Struct(fields) = calendar_interval_type() else {
+            unreachable!()
+        };
+        Ok(ColumnarValue::Array(Arc::new(StructArray::new(
+            fields,
+            columns,
+            Some(NullBuffer::from(valid)),
+        ))))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn preserves_spark_microsecond_range_and_overflow() {
+        assert_eq!(
+            make_interval(1, 2, 3, 4, 2_562_048, 0, 123_456_789_012_123_456),
+            Ok((14, 25, 132_680_161_812_123_456))
+        );
+        assert_eq!(make_interval(i32::MAX, 0, 0, 0, 0, 0, 0), Err("integer"));
+        assert_eq!(
+            make_interval(0, 0, 0, 0, i32::MAX, i32::MAX, i128::MAX),
+            Err("long")
+        );
     }
 }

native/spark-expr/src/datetime_funcs/mod.rs

@@ -39,7 +39,7 @@ pub use extract_date_part::SparkMinute;
 pub use extract_date_part::SparkSecond;
 pub use hours::SparkHoursTransform;
 pub use make_date::SparkMakeDate;
-pub use make_interval::SparkMakeInterval;
+pub use make_interval::{calendar_interval_type, SparkMakeInterval};
 pub use make_time::SparkMakeTime;
 pub use next_day::SparkNextDay;
 pub use seconds_to_timestamp::SparkSecondsToTimestamp;

native/spark-expr/src/lib.rs

@@ -76,10 +76,10 @@ pub use comet_scalar_funcs::{
 };
 pub use csv_funcs::*;
 pub use datetime_funcs::{
-    spark_day_name, spark_month_name, spark_to_time, SparkDateDiff, SparkDateFromUnixDate,
-    SparkDateTrunc, SparkHour, SparkHoursTransform, SparkMakeDate, SparkMakeInterval,
-    SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond, SparkSecondsToTimestamp,
-    SparkUnixTimestamp, TimestampTruncExpr,
+    calendar_interval_type, spark_day_name, spark_month_name, spark_to_time, SparkDateDiff,
+    SparkDateFromUnixDate, SparkDateTrunc, SparkHour, SparkHoursTransform, SparkMakeDate,
+    SparkMakeInterval, SparkMakeTime, SparkMinute, SparkNextDay, SparkSecond,
+    SparkSecondsToTimestamp, SparkUnixTimestamp, TimestampTruncExpr,
 };
 pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult};
 pub use hash_funcs::*;