diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 4ad4d9afe3..24b8f07bf7 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -401,7 +401,7 @@ expression-level). The `outer` variants are wired but marked `Incompatible`; the | `*` | ✅ | Interval multiplication falls back | | `+` | ✅ | | | `-` | ✅ | | -| `/` | ✅ | | +| `/` | ✅ | DayTime interval division routes through the JVM codegen dispatcher; YearMonth and Calendar interval division fall back | | `abs` | ✅ | Interval types fall back | | `acos` | ✅ | | | `acosh` | ✅ | | diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 7ed2b3331c..dff27fc95c 100644 --- a/native/core/src/execution/planner.rs +++ b/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:?}" ))) } }, 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 09bfc52bd4..55c52af1e6 100644 --- a/spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegenInput.scala +++ b/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 { 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 5eee0c5cad..502c678d3e 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -284,6 +284,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { classOf[ConvertTimezone] -> CometConvertTimezone, classOf[DateAdd] -> CometDateAdd, classOf[DateDiff] -> CometDateDiff, + classOf[DivideDTInterval] -> CometDivideDTInterval, classOf[DateFormatClass] -> CometDateFormat, classOf[DateFromUnixDate] -> CometDateFromUnixDate, classOf[Days] -> CometDays, 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 9f2533d85f..8542904d08 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, DivideDTInterval, 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.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 CometDivideDTInterval extends CometCodegenDispatch[DivideDTInterval] + /** * Spark's internal `PreciseTimestampConversion` reinterprets a value between the timestamp types * (`TimestampType` / `TimestampNTZType`) and `LongType` without losing microsecond precision. It 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 4f2a5dfa5e..bac3631bab 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, 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]) 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 f575dd5b53..145698102e 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 @@ -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 => diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/divide_dt_interval.sql b/spark/src/test/resources/sql-tests/expressions/datetime/divide_dt_interval.sql new file mode 100644 index 0000000000..c7b16f8aa2 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/datetime/divide_dt_interval.sql @@ -0,0 +1,65 @@ +-- 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 divide_dt_interval through the codegen dispatcher; produces DayTimeIntervalType. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true + +statement +CREATE TABLE test_divide_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_divide_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(2 AS TINYINT), CAST(2 AS SMALLINT), 2, CAST(2 AS BIGINT), CAST(2.0 AS FLOAT), CAST(2.0 AS DOUBLE), CAST(2.00 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_divide_dt_interval + +-- literal interval input +query +SELECT INTERVAL '1 02:03:04.500000' DAY TO SECOND / i FROM test_divide_dt_interval + +-- literal divisors, including half-up rounding to the nearest microsecond. +query +SELECT + make_dt_interval(1, 2, 3, 4.5) / 2, + INTERVAL '0.000001' SECOND / 2, + INTERVAL '0.000001' SECOND / CAST(2.00 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_divide_dt_interval + +-- Division by zero fails regardless of ANSI mode. +query expect_error(zero) +SELECT make_dt_interval(1) / 0 + +-- This interval is Long.MinValue microseconds, which cannot be divided by -1. +query expect_error(overflow) +SELECT make_dt_interval(-106751991, -4, 0, -54.775808) / -1