feat: add native make_interval support #5039
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| // 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 Default for SparkMakeInterval { | ||
| fn default() -> Self { | ||
| Self::new(false) | ||
| } | ||
| } | ||
|
|
||
| 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)?; | ||
|
Member
There was a problem hiding this comment. There's a compatibility concern I'd like to flag with the underlying DataFusion kernel. Two related issues: Nanosecond vs microsecond overflow. Spark's select make_interval(1, 2, 3, 4, 0, 0, 123456789012.123456);Float64 coercion loses microsecond precision. DataFusion's Given both, would it make sense to mark this expression |
||
|
|
||
| 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 { | ||
| return Err(arithmetic_overflow_error("interval").into()); | ||
| } | ||
| } | ||
|
|
||
| Ok(result) | ||
| } | ||
| } | ||
| 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[MicrosToTimestamp] -> CometMicrosToTimestamp, | ||
| classOf[MillisToTimestamp] -> CometMillisToTimestamp, | ||
|
|
@@ -1111,9 +1112,16 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { | |
| } | ||
|
|
||
| def scalarFunctionExprToProto(funcName: String, args: Option[Expr]*): Option[Expr] = { | ||
| scalarFunctionExprToProto(funcName, false, args: _*) | ||
| } | ||
|
|
||
| def scalarFunctionExprToProto( | ||
| funcName: String, | ||
| failOnError: Boolean, | ||
| args: Option[Expr]*): Option[Expr] = { | ||
| val builder = ExprOuterClass.ScalarFunc.newBuilder() | ||
| builder.setFunc(funcName) | ||
| builder.setFailOnError(failOnError) | ||
| scalarFunctionExprToProto0(builder, args: _*) | ||
| } | ||
|
|
||
|
|
||
| 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, 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 | ||
|
|
@@ -968,6 +968,33 @@ object CometMakeYMInterval extends CometCodegenDispatch[MakeYMInterval] | |
|
|
||
| object CometMakeDTInterval extends CometCodegenDispatch[MakeDTInterval] | ||
|
|
||
| object CometMakeInterval extends CometExpressionSerde[MakeInterval] { | ||
|
Member
There was a problem hiding this comment. One structural thought. Since Would you be open to mixing in object CometMakeInterval
extends CometExpressionSerde[MakeInterval]
with CodegenDispatchFallback {That routes the non-opt-in This would also let #5260 and this PR land together rather than one replacing the other. I opened #5260 as the codegen-dispatch route before seeing how far this one had come. If you would rather keep them separate I am happy to close #5260 and let you carry the dispatch mixin here.
Member
Author
There was a problem hiding this comment. Thanks for the direction! I'd like to take the mixin approach. feel free to close #5260. |
||
| private val incompatReason = | ||
| "The native implementation converts seconds to `Float64`, which can lose microsecond" + | ||
| " precision, and stores time in nanoseconds, which overflows for large seconds values" + | ||
| " that Spark can represent." | ||
|
|
||
|
Comment on lines
+972
to
+976
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.children(6), DoubleType)) | ||
| val childExprs = children.map(exprToProtoInternal(_, inputs, binding)) | ||
| val optExpr = scalarFunctionExprToProtoWithReturnType( | ||
| "make_interval", | ||
| CalendarIntervalType, | ||
| expr.failOnError, | ||
| childExprs: _*) | ||
| optExprWithFallbackReason(optExpr, expr, children: _*) | ||
| } | ||
| } | ||
|
|
||
| object CometMultiplyDTInterval extends CometCodegenDispatch[MultiplyDTInterval] | ||
|
|
||
| /** | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| -- 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 | ||
| 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) |
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| -- 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) | ||
| 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) | ||
| SELECT make_interval(0, 0, 2147483647) | ||
|
Comment on lines
+34
to
+38
Member
There was a problem hiding this comment. Would you tighten these to Note the message bodies still differ. Spark says |
||
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.
I do not think this registration is reachable. Because the serde sets the return type,
create_scalar_function_exprskips thesession_ctx.udf(...)lookup (planner.rs:3337) and goes tocreate_comet_physical_fun, where the"make_interval"arm buildsSparkMakeInterval::new(fail_on_error)directly.If you want to keep a registry entry for safety,
SparkMakeDateis the precedent and it lives inall_scalar_functions()incomet_scalar_funcs.rs. Putting a Comet wrapper here is a little misleading, since everything else inregister_datafusion_spark_functionis a raw upstream UDF, and this one hardcodesfail_on_error = falseviaDefault. If it ever did get used it would silently ignore ANSI.