ARROW-12277: [Rust][DataFusion] Implement and test Count/Min/Max aggregates for Timestamp(_,_)

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }

rust/datafusion/src/execution/context.rs

@@ -1403,6 +1403,121 @@ mod tests {
  Ok(())
  }
 
+ #[tokio::test]
+ async fn aggregate_timestamps_sum() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT sum(nanos), sum(micros), sum(millis), sum(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_count() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT count(nanos), count(micros), count(millis), count(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+--------------+---------------+---------------+-------------+",
+ "| COUNT(nanos) | COUNT(micros) | COUNT(millis) | COUNT(secs) |",
+ "+--------------+---------------+---------------+-------------+",
+ "| 3 | 3 | 3 | 3 |",
+ "+--------------+---------------+---------------+-------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_min() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT min(nanos), min(micros), min(millis), min(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| MIN(nanos) | MIN(micros) | MIN(millis) | MIN(secs) |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ "| 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123450 | 2011-12-13 11:13:10.123 | 2011-12-13 11:13:10 |",
+ "+----------------------------+----------------------------+-------------------------+---------------------+",
+ ];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_max() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT max(nanos), max(micros), max(millis), max(secs) FROM t",
+ )
+ .await
+ .unwrap();
+
+ let expected = vec![
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| MAX(nanos) | MAX(micros) | MAX(millis) | MAX(secs) |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+ "| 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10.432 | 2021-01-01 05:11:10 |",
+ "+-------------------------+-------------------------+-------------------------+---------------------+",
+];
+ assert_batches_sorted_eq!(expected, &results);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn aggregate_timestamps_avg() -> Result<()> {
+ let tmp_dir = TempDir::new()?;
+ let mut ctx = create_ctx(&tmp_dir, 1)?;
+ ctx.register_table("t", test::table_with_timestamps())
+ .unwrap();
+
+ let results = plan_and_collect(
+ &mut ctx,
+ "SELECT avg(nanos), avg(micros), avg(millis), avg(secs) FROM t",
+ )
+ .await
+ .unwrap_err();
+
+ assert_eq!(results.to_string(), "Error during planning: Coercion from [Timestamp(Nanosecond, None)] to the signature Uniform(1, [Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64, Float32, Float64]) failed.");
+ Ok(())
+ }
+
  #[tokio::test]
  async fn join_partitioned() -> Result<()> {
  // self join on partition id (workaround for duplicate column name)

rust/datafusion/src/physical_plan/aggregates.rs

@@ -34,7 +34,7 @@ use super::{
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::distinct_expressions;
 use crate::physical_plan::expressions;
-use arrow::datatypes::{DataType, Schema};
+use arrow::datatypes::{DataType, Schema, TimeUnit};
 use expressions::{avg_return_type, sum_return_type};
 use std::{fmt, str::FromStr, sync::Arc};
 
@@ -160,6 +160,8 @@ pub fn create_aggregate_expr(
  })
 }
 
+static STRINGS: &[DataType] = &[DataType::Utf8, DataType::LargeUtf8];
+
 static NUMERICS: &[DataType] = &[
  DataType::Int8,
  DataType::Int16,
@@ -173,14 +175,25 @@ static NUMERICS: &[DataType] = &[
  DataType::Float64,
 ];
 
+static TIMESTAMPS: &[DataType] = &[
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+];
+
 /// the signatures supported by the function `fun`.
 fn signature(fun: &AggregateFunction) -> Signature {
  // note: the physical expression must accept the type returned by this function or the execution panics.
  match fun {
  AggregateFunction::Count => Signature::Any(1),
  AggregateFunction::Min | AggregateFunction::Max => {
- let mut valid = vec![DataType::Utf8, DataType::LargeUtf8];
- valid.extend_from_slice(NUMERICS);
+ let valid = STRINGS
+ .iter()
+ .chain(NUMERICS.iter())
+ .chain(TIMESTAMPS.iter())
+ .cloned()
+ .collect::<Vec<_>>();
  Signature::Uniform(1, valid)
  }
  AggregateFunction::Avg | AggregateFunction::Sum => {

rust/datafusion/src/physical_plan/datetime_expressions.rs

@@ -324,8 +324,8 @@ pub fn date_trunc(args: &[ColumnarValue]) -> Result<ColumnarValue> {
 
  Ok(match array {
  ColumnarValue::Scalar(scalar) => {
- if let ScalarValue::TimeNanosecond(v) = scalar {
- ColumnarValue::Scalar(ScalarValue::TimeNanosecond((f)(*v)?))
+ if let ScalarValue::TimestampNanosecond(v) = scalar {
+ ColumnarValue::Scalar(ScalarValue::TimestampNanosecond((f)(*v)?))
  } else {
  return Err(DataFusionError::Execution(
  "array of `date_trunc` must be non-null scalar Utf8".to_string(),

rust/datafusion/src/physical_plan/expressions/min_max.rs

@@ -25,12 +25,13 @@ use crate::error::{DataFusionError, Result};
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use arrow::compute;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, TimeUnit};
 use arrow::{
  array::{
  ArrayRef, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
- Int8Array, LargeStringArray, StringArray, UInt16Array, UInt32Array, UInt64Array,
- UInt8Array,
+ Int8Array, LargeStringArray, StringArray, TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ UInt16Array, UInt32Array, UInt64Array, UInt8Array,
  },
  datatypes::Field,
 };
@@ -128,6 +129,27 @@ macro_rules! min_max_batch {
  DataType::UInt32 => typed_min_max_batch!($VALUES, UInt32Array, UInt32, $OP),
  DataType::UInt16 => typed_min_max_batch!($VALUES, UInt16Array, UInt16, $OP),
  DataType::UInt8 => typed_min_max_batch!($VALUES, UInt8Array, UInt8, $OP),
+ DataType::Timestamp(TimeUnit::Second, _) => {
+ typed_min_max_batch!($VALUES, TimestampSecondArray, TimestampSecond, $OP)
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMillisecondArray,
+ TimestampMillisecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Microsecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampMicrosecondArray,
+ TimestampMicrosecond,
+ $OP
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => typed_min_max_batch!(
+ $VALUES,
+ TimestampNanosecondArray,
+ TimestampNanosecond,
+ $OP
+ ),
  other => {
  // This should have been handled before
  return Err(DataFusionError::Internal(format!(
@@ -229,6 +251,27 @@ macro_rules! min_max {
  (ScalarValue::LargeUtf8(lhs), ScalarValue::LargeUtf8(rhs)) => {
  typed_min_max_string!(lhs, rhs, LargeUtf8, $OP)
  }
+ (ScalarValue::TimestampSecond(lhs), ScalarValue::TimestampSecond(rhs)) => {
+ typed_min_max!(lhs, rhs, TimestampSecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMillisecond(lhs),
+ ScalarValue::TimestampMillisecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMillisecond, $OP)
+ }
+ (
+ ScalarValue::TimestampMicrosecond(lhs),
+ ScalarValue::TimestampMicrosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampMicrosecond, $OP)
+ }
+ (
+ ScalarValue::TimestampNanosecond(lhs),
+ ScalarValue::TimestampNanosecond(rhs),
+ ) => {
+ typed_min_max!(lhs, rhs, TimestampNanosecond, $OP)
+ }
  e => {
  return Err(DataFusionError::Internal(format!(
  "MIN/MAX is not expected to receive a scalar {:?}",

rust/datafusion/src/physical_plan/group_scalar.rs

@@ -64,9 +64,15 @@ impl TryFrom<&ScalarValue> for GroupByScalar {
  ScalarValue::UInt16(Some(v)) => GroupByScalar::UInt16(*v),
  ScalarValue::UInt32(Some(v)) => GroupByScalar::UInt32(*v),
  ScalarValue::UInt64(Some(v)) => GroupByScalar::UInt64(*v),
- ScalarValue::TimeMillisecond(Some(v)) => GroupByScalar::TimeMillisecond(*v),
- ScalarValue::TimeMicrosecond(Some(v)) => GroupByScalar::TimeMicrosecond(*v),
- ScalarValue::TimeNanosecond(Some(v)) => GroupByScalar::TimeNanosecond(*v),
+ ScalarValue::TimestampMillisecond(Some(v)) => {
+ GroupByScalar::TimeMillisecond(*v)
+ }
+ ScalarValue::TimestampMicrosecond(Some(v)) => {
+ GroupByScalar::TimeMicrosecond(*v)
+ }
+ ScalarValue::TimestampNanosecond(Some(v)) => {
+ GroupByScalar::TimeNanosecond(*v)
+ }
  ScalarValue::Utf8(Some(v)) => GroupByScalar::Utf8(Box::new(v.clone())),
  ScalarValue::Float32(None)
  | ScalarValue::Float64(None)
@@ -110,9 +116,15 @@ impl From<&GroupByScalar> for ScalarValue {
  GroupByScalar::UInt32(v) => ScalarValue::UInt32(Some(*v)),
  GroupByScalar::UInt64(v) => ScalarValue::UInt64(Some(*v)),
  GroupByScalar::Utf8(v) => ScalarValue::Utf8(Some(v.to_string())),
- GroupByScalar::TimeMillisecond(v) => ScalarValue::TimeMillisecond(Some(*v)),
- GroupByScalar::TimeMicrosecond(v) => ScalarValue::TimeMicrosecond(Some(*v)),
- GroupByScalar::TimeNanosecond(v) => ScalarValue::TimeNanosecond(Some(*v)),
+ GroupByScalar::TimeMillisecond(v) => {
+ ScalarValue::TimestampMillisecond(Some(*v))
+ }
+ GroupByScalar::TimeMicrosecond(v) => {
+ ScalarValue::TimestampMicrosecond(Some(*v))
+ }
+ GroupByScalar::TimeNanosecond(v) => {
+ ScalarValue::TimestampNanosecond(Some(*v))
+ }
  GroupByScalar::Date32(v) => ScalarValue::Date32(Some(*v)),
  }
  }