feat: add make_interval support (codegen dispatch + native)
#5039
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| // 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. | ||
|
|
||
| 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; | ||
|
|
||
| #[derive(Debug, PartialEq, Eq, Hash)] | ||
| pub struct SparkMakeInterval { | ||
| inner: DataFusionMakeInterval, | ||
| fail_on_error: bool, | ||
| } | ||
|
|
||
| impl SparkMakeInterval { | ||
| pub fn new(fail_on_error: bool) -> Self { | ||
| Self { | ||
| inner: DataFusionMakeInterval::new(), | ||
| fail_on_error, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl ScalarUDFImpl for SparkMakeInterval { | ||
| fn name(&self) -> &str { | ||
| self.inner.name() | ||
| } | ||
|
|
||
| fn signature(&self) -> &Signature { | ||
| self.inner.signature() | ||
| } | ||
|
|
||
| fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> { | ||
| self.inner.return_type(arg_types) | ||
| } | ||
|
|
||
| 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()); | ||
| } | ||
| } | ||
|
|
||
| Ok(result) | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -303,6 +303,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { | |
| classOf[MakeTimestamp] -> CometMakeTimestamp, | ||
| classOf[MakeYMInterval] -> CometMakeYMInterval, | ||
| classOf[MakeDTInterval] -> CometMakeDTInterval, | ||
| classOf[MakeInterval] -> CometMakeInterval, | ||
|
Member
There was a problem hiding this comment.
Two follow-ons. The |
||
| classOf[MultiplyDTInterval] -> CometMultiplyDTInterval, | ||
| classOf[TimestampAdd] -> CometTimestampAdd, | ||
| classOf[TimestampDiff] -> CometTimestampDiff, | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,9 +21,9 @@ package org.apache.comet.serde | |
|
|
||
| import java.util.Locale | ||
|
|
||
| import org.apache.spark.sql.catalyst.expressions.{AddMonths, Attribute, Cast, ConvertTimezone, DateAdd, DateDiff, DateFormatClass, DateFromUnixDate, DateSub, DayOfMonth, DayOfWeek, DayOfYear, Days, Expression, FromUTCTimestamp, GetDateField, GetTimestamp, Hour, Hours, LastDay, Literal, MakeDate, MakeDTInterval, MakeInterval, MakeTimestamp, MakeYMInterval, MicrosToTimestamp, MillisToTimestamp, Minute, Month, MonthsBetween, MultiplyDTInterval, NextDay, PreciseTimestampConversion, Quarter, Second, SecondsToTimestamp, TimestampAdd, TimestampDiff, 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.{CalendarIntervalType, DataType, DateType, DoubleType, FloatType, IntegerType, LongType, StringType, TimestampNTZType, TimestampType} | ||
| import org.apache.spark.unsafe.types.UTF8String | ||
|
|
||
| import org.apache.comet.CometConf | ||
|
|
@@ -963,6 +963,39 @@ object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval] | |
|
|
||
| object CometMakeDTInterval extends CometCodegenDispatch[MakeDTInterval] | ||
|
|
||
| object CometMakeInterval extends CometExpressionSerde[MakeInterval] with CodegenDispatchFallback { | ||
|
Member
There was a problem hiding this comment. The default dispatch path has a range limit of its own, and I do not think the checkmark in I built this branch and ran The source is This is a constraint of Comet's |
||
| private val incompatReason = | ||
| "The native implementation converts seconds to `Float64`, which can lose microsecond" + | ||
| " precision, and stores time in nanoseconds, which overflows for large time components" + | ||
| " (hours, minutes, seconds) that Spark can represent." | ||
|
|
||
| override def getCompatibleNotes(): Seq[String] = Seq( | ||
| "Both the default JVM codegen-dispatch path and the native path currently limit the" + | ||
| " elapsed-time component to about 292 years in either direction. This only affects" + | ||
| " extreme intervals and is tracked in" + | ||
| " [#5279](https://github.com/apache/datafusion-comet/issues/5279).") | ||
|
|
||
|
Comment on lines
+967
to
+971
Member
There was a problem hiding this comment. The reason attributes the nanosecond overflow to seconds, but hours and minutes hit it too, and at much lower values. Spark's
Could the reason say "time components (hours, minutes, seconds)" rather than just seconds? This string is what renders on the generated compat page, so it is the only warning a user gets. It would be good to widen #5131's description the same way. |
||
| override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason) | ||
|
|
||
| override def getSupportLevel(expr: MakeInterval): SupportLevel = | ||
| Incompatible(Some(incompatReason)) | ||
|
|
||
| override def convert( | ||
| expr: MakeInterval, | ||
| inputs: Seq[Attribute], | ||
| binding: Boolean): Option[Expr] = { | ||
| // The explicit return type skips DataFusion's registry coercion, but its kernel needs Float64. | ||
| val children = expr.children.updated(6, Cast(expr.secs, DoubleType)) | ||
| val childExprs = children.map(exprToProtoInternal(_, inputs, binding)) | ||
| val optExpr = scalarFunctionExprToProtoWithReturnType( | ||
| "make_interval", | ||
| CalendarIntervalType, | ||
| expr.failOnError, | ||
| childExprs: _*) | ||
| optExpr | ||
| } | ||
| } | ||
|
|
||
| object CometMultiplyDTInterval extends CometCodegenDispatch[MultiplyDTInterval] | ||
|
|
||
| object CometTimestampAdd extends CometCodegenDispatch[TimestampAdd] | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| -- Licensed to the Apache Software Foundation (ASF) under one | ||
| -- or more contributor license agreements. See the NOTICE file | ||
| -- distributed with this work for additional information | ||
| -- regarding copyright ownership. The ASF licenses this file | ||
| -- to you under the Apache License, Version 2.0 (the | ||
| -- "License"); you may not use this file except in compliance | ||
| -- with the License. You may obtain a copy of the License at | ||
| -- | ||
| -- http://www.apache.org/licenses/LICENSE-2.0 | ||
| -- | ||
| -- Unless required by applicable law or agreed to in writing, | ||
| -- software distributed under the License is distributed on an | ||
| -- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| -- KIND, either express or implied. See the License for the | ||
| -- specific language governing permissions and limitations | ||
| -- under the License. | ||
|
|
||
| -- Config: spark.comet.expression.MakeInterval.allowIncompatible=true | ||
|
|
||
| statement | ||
| CREATE TABLE test_make_interval( | ||
| years int, | ||
| months int, | ||
| weeks int, | ||
| days int, | ||
| hours int, | ||
| mins int, | ||
| secs decimal(18, 6)) USING parquet | ||
|
|
||
| statement | ||
| INSERT INTO test_make_interval VALUES | ||
| (1, 2, 3, 4, 5, 6, 7.123456), | ||
| (0, 1, 0, 1, 0, 0, 100.000001), | ||
| (-1, -2, -1, -1, -1, -1, -1.500000), | ||
| (NULL, 1, 2, 3, 4, 5, 6.000000), | ||
| (2, NULL, 2, 3, 4, 5, 6.000000), | ||
| (3, 1, 2, 3, 4, 5, NULL), | ||
| (-2147483648, 0, 0, 0, 0, 0, 0.000000) | ||
|
|
||
| query | ||
| SELECT make_interval(years, months, weeks, days, hours, mins, secs) | ||
| FROM test_make_interval | ||
| ORDER BY years | ||
|
|
||
| query | ||
| SELECT make_interval(1, 2), make_interval(3), make_interval() | ||
|
|
||
| query | ||
| SELECT make_interval(0, 1, 0, 1, 0, 0, 100.000001) | ||
|
|
||
| query | ||
| SELECT make_interval(2147483647) | ||
|
Member
There was a problem hiding this comment. It might be worth expanding coverage here. A few things Spark's own
|
||
|
|
||
| query ignore(https://github.com/apache/datafusion-comet/issues/5131) | ||
| SELECT make_interval(1, 2, 3, 4, 0, 0, 123456789012.123456) | ||
|
|
||
| query | ||
| SELECT make_interval(0, 0, 0, 0, 0, 0, 999999999.999999) | ||
|
|
||
| query ignore(https://github.com/apache/datafusion-comet/issues/5131) | ||
| SELECT make_interval(0, 0, 0, 0, 0, 0, 999999999.000001) | ||
|
|
||
| query ignore(https://github.com/apache/datafusion-comet/issues/5131) | ||
| SELECT make_interval(0, 0, 0, 0, 2562048) | ||
|
|
||
| query | ||
| SELECT make_interval(0, 0, 0, 0, 0, 0, 1234567890123456789) | ||
|
Member
There was a problem hiding this comment. Could you add an hours case alongside the seconds ones? It is the same #5131 nanosecond overflow but on a component the fixture does not touch, and at a value a real query is much more likely to produce than a 12-digit seconds decimal. query ignore(https://github.com/apache/datafusion-comet/issues/5131)
SELECT make_interval(0, 0, 0, 0, 2562048) |
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| -- 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. | ||
|
|
||
| -- Native ANSI execution must preserve Spark's overflow exception. | ||
| -- Config: spark.sql.ansi.enabled=true | ||
| -- Config: spark.comet.expression.MakeInterval.allowIncompatible=true | ||
|
|
||
| statement | ||
| CREATE TABLE test_make_interval_ansi(years int) USING parquet | ||
|
|
||
| statement | ||
| INSERT INTO test_make_interval_ansi VALUES (NULL) | ||
|
|
||
| query | ||
| SELECT make_interval(1, 2, 3, 4, 5, 6, 7.123456) | ||
|
|
||
| query | ||
| SELECT make_interval(years) FROM test_make_interval_ansi | ||
|
|
||
| query expect_error(overflow. If necessary set) | ||
| SELECT make_interval(2147483647) | ||
|
Member
There was a problem hiding this comment. Consider adding ANSI overflow tests for components other than query expect_error(overflow)
SELECT make_interval(0, 0, 2147483647)would confirm the overflow detection path fires on non- |
||
|
|
||
| query expect_error(overflow. If necessary set) | ||
| SELECT make_interval(0, 0, 2147483647) | ||
|
|
||
| query ignore(https://github.com/apache/datafusion-comet/issues/5131) | ||
| SELECT make_interval(0, 0, 0, 0, 2562048) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| -- 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. | ||
|
|
||
| -- With allowIncompatible unset, MakeInterval uses Spark's JVM codegen dispatcher. | ||
|
|
||
| statement | ||
| CREATE TABLE test_make_interval_dispatch( | ||
| years int, | ||
| months int, | ||
| weeks int, | ||
| days int, | ||
| hours int, | ||
| mins int, | ||
| secs decimal(18, 6)) USING parquet | ||
|
|
||
| statement | ||
| INSERT INTO test_make_interval_dispatch VALUES | ||
| (1, 2, 3, 4, 5, 6, 7.123456), | ||
| (0, 1, 0, 1, 0, 0, 100.000001), | ||
| (-1, -2, -1, -1, -1, -1, -1.500000), | ||
| (NULL, 1, 2, 3, 4, 5, 6.000000), | ||
| (2, NULL, 2, 3, 4, 5, 6.000000), | ||
| (3, 1, 2, 3, 4, 5, NULL), | ||
| (0, 0, 0, 0, 2562048, 0, 0.000000) | ||
|
|
||
| query | ||
| SELECT make_interval(years, months, weeks, days, hours, mins, secs) | ||
| FROM test_make_interval_dispatch | ||
| WHERE hours != 2562048 | ||
| ORDER BY years | ||
|
|
||
| query ignore(https://github.com/apache/datafusion-comet/issues/5279) | ||
| SELECT make_interval(0, 0, 0, 0, hours) | ||
| FROM test_make_interval_dispatch | ||
| WHERE hours = 2562048 |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There's a compatibility concern I'd like to flag with the underlying DataFusion kernel. Two related issues:
Nanosecond vs microsecond overflow. Spark's
IntervalUtils.makeIntervalstores time components asint64microseconds viasecs.toUnscaledLong, so aDecimal(18, 6)seconds value fits comfortably (max ≈ 1e18 micros, well underLong.MaxValue). DataFusion's kernel accumulates in nanoseconds, so it overflows at roughlysecs > 9_223_372_036(~292 years). Any Decimal(18, 6) seconds value beyond that boundary silently returns null under this PR (or throws under ANSI) while Spark returns a valid interval. Spark's ownsql-tests/inputs/interval.sqlexercises exactly this range:Float64 coercion loses microsecond precision. DataFusion's
SparkMakeIntervalsignature coercessecstoFloat64, but Spark'sMakeInterval.inputTypesisDecimal(18, 6)and preserves microseconds exactly. Forsecs = 999999999.999999, the Float64 round-trip yieldsfrac * 1e9 ≈ 999999046instead of999999000— a ~46 ns drift that translates into a wrong microsecond count on the JVM side. The small values currently in the fixture (7.123456,100.000001,-1.5) happen to be exactly representable so they don't expose this.Given both, would it make sense to mark this expression
Incompatible(Some("..."))ingetSupportLevel, and add agetIncompatibleReasons()string so the auto-generated compat page warns users? Marking itNativeinexpressions.mdwith no caveat currently overstates the compatibility.