martin-g commented on code in PR #19341:
URL: https://github.com/apache/datafusion/pull/19341#discussion_r2623042483
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32MillisecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time32(Second) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32SecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32SecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Microsecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
Review Comment:
This is already imported at line 27 as `use arrow::array::AsArray;`
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32MillisecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time32(Second) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32SecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32SecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Microsecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
+ let array = array.as_primitive::<Time64MicrosecondType>();
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
+ binned_nanos % (ONE_DAY_IN_NS)
+ };
+ let array: PrimitiveArray<Time64MicrosecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Nanosecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
Review Comment:
It is already imported at line 27 as `use arrow::array::AsArray;`
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32MillisecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time32(Second) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32SecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32SecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Microsecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
+ let array = array.as_primitive::<Time64MicrosecondType>();
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
+ binned_nanos % (ONE_DAY_IN_NS)
+ };
+ let array: PrimitiveArray<Time64MicrosecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Nanosecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
+ let array = array.as_primitive::<Time64NanosecondType>();
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ nanos / 1_000
Review Comment:
```suggestion
nanos
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32MillisecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time32(Second) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32SecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32SecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time64(Microsecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time64 source requires Time64
origin"
+ );
+ }
+ use arrow::array::cast::AsArray;
+ let array = array.as_primitive::<Time64MicrosecondType>();
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
+ binned_nanos % (ONE_DAY_IN_NS)
Review Comment:
```suggestion
binned_nanos % (ONE_DAY_IN_NS) / 1_000
```
to convert to microseconds
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -439,7 +530,49 @@ fn date_bin_impl(
tz_opt.clone(),
))
}
-
+ ColumnarValue::Scalar(ScalarValue::Time32Millisecond(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time32 source requires Time32
origin");
+ }
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+
ColumnarValue::Scalar(ScalarValue::Time32Millisecond(v.map(apply_stride_fn)))
+ }
+ ColumnarValue::Scalar(ScalarValue::Time32Second(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time32 source requires Time32
origin");
+ }
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
Review Comment:
`x` must be converted to nanos
```suggestion
let binned_nanos = stride_fn(stride, x as i64 *
1_000_000_000, origin);
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+ let array: PrimitiveArray<Time32MillisecondType> =
+ array.unary(apply_stride_fn);
+ ColumnarValue::Array(Arc::new(array))
+ }
+ Time32(Second) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32SecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
Review Comment:
```suggestion
let binned_nanos = stride_fn(stride, x as i64 *
1_000_000_000, origin);
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -481,17 +614,82 @@ fn date_bin_impl(
origin, stride, stride_fn, array, tz_opt,
)?
}
+ Time32(Millisecond) => {
+ if !is_time {
+ return exec_err!(
+ "DATE_BIN with Time32 source requires Time32
origin"
+ );
+ }
+ let array = array.as_primitive::<Time32MillisecondType>();
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
Review Comment:
`x` must be converted to nanos
```suggestion
let binned_nanos = stride_fn(stride, x as i64 * 1_000_000,
origin);
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -439,7 +530,49 @@ fn date_bin_impl(
tz_opt.clone(),
))
}
-
+ ColumnarValue::Scalar(ScalarValue::Time32Millisecond(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time32 source requires Time32
origin");
+ }
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000) as i32
+ };
+
ColumnarValue::Scalar(ScalarValue::Time32Millisecond(v.map(apply_stride_fn)))
+ }
+ ColumnarValue::Scalar(ScalarValue::Time32Second(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time32 source requires Time32
origin");
+ }
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
+ let nanos = binned_nanos % (ONE_DAY_IN_NS);
+ (nanos / 1_000_000_000) as i32
+ };
+
ColumnarValue::Scalar(ScalarValue::Time32Second(v.map(apply_stride_fn)))
+ }
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time64 source requires Time64
origin");
+ }
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
+ binned_nanos % (ONE_DAY_IN_NS)
+ };
+
ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(v.map(apply_stride_fn)))
+ }
+ ColumnarValue::Scalar(ScalarValue::Time64Microsecond(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time64 source requires Time64
origin");
+ }
+ let apply_stride_fn = move |x: i64| {
+ let binned_nanos = stride_fn(stride, x, origin);
Review Comment:
`x` must be converted to nanos
```suggestion
let binned_nanos = stride_fn(stride, x as i64 * 1_000,
origin);
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -439,7 +530,49 @@ fn date_bin_impl(
tz_opt.clone(),
))
}
-
+ ColumnarValue::Scalar(ScalarValue::Time32Millisecond(v)) => {
+ if !is_time {
+ return exec_err!("DATE_BIN with Time32 source requires Time32
origin");
+ }
+ let apply_stride_fn = move |x: i32| {
+ let binned_nanos = stride_fn(stride, x as i64, origin);
Review Comment:
`x` must be converted to nanos
```suggestion
let binned_nanos = stride_fn(stride, x as i64 * 1_000_000,
origin);
```
##########
datafusion/functions/src/datetime/date_bin.rs:
##########
@@ -370,11 +419,53 @@ fn date_bin_impl(
}
};
- let origin = match origin {
- ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(Some(v), _)) =>
*v,
+ let (origin, is_time) = match origin {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(Some(v), _)) =>
{
+ (*v, false)
+ }
+ ColumnarValue::Scalar(ScalarValue::Time32Second(Some(v))) => {
+ match stride {
+ Interval::Months(m) => {
+ if m > 0 {
+ return exec_err!(
+ "DATE_BIN stride for TIME input must be less than
1 day"
+ );
+ }
+ }
+ Interval::Nanoseconds(ns) => {
+ if ns >= ONE_DAY_IN_NS {
+ return exec_err!(
+ "DATE_BIN stride for TIME input must be less than
1 day"
+ );
+ }
+ }
+ }
+
+ (*v as i64, true)
+ }
+ ColumnarValue::Scalar(ScalarValue::Time64Nanosecond(Some(v))) => {
+ match stride {
+ Interval::Months(m) => {
+ if m > 0 {
+ return exec_err!(
+ "DATE_BIN stride for TIME input must be less than
1 day"
+ );
+ }
+ }
+ Interval::Nanoseconds(ns) => {
+ if ns >= ONE_DAY_IN_NS {
+ return exec_err!(
+ "DATE_BIN stride for TIME input must be less than
1 day"
+ );
+ }
+ }
+ }
+
+ (*v, true)
+ }
ColumnarValue::Scalar(v) => {
Review Comment:
Shouldn't there be also arms for `Time32(Millisecond)` and
`Time64(Microsecond)` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
