diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs index 6720152cbd3a7..6e6ee1ba183e7 100644 --- a/datafusion/functions/src/datetime/to_timestamp.rs +++ b/datafusion/functions/src/datetime/to_timestamp.rs @@ -31,6 +31,7 @@ use arrow::datatypes::{ }; use datafusion_common::config::ConfigOptions; use datafusion_common::{Result, ScalarType, ScalarValue, exec_datafusion_err, exec_err}; +use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use datafusion_expr::{ ColumnarValue, Documentation, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, @@ -581,6 +582,16 @@ impl ScalarUDFImpl for ToTimestampSecondsFunc { } } + fn output_ordering(&self, input: &[ExprProperties]) -> Result { + if let [value] = input + && value.range.data_type().is_integer() + { + Ok(value.sort_properties) + } else { + Ok(SortProperties::Unordered) + } + } + fn documentation(&self) -> Option<&Documentation> { self.doc() } @@ -650,6 +661,16 @@ impl ScalarUDFImpl for ToTimestampMillisFunc { } } + fn output_ordering(&self, input: &[ExprProperties]) -> Result { + if let [value] = input + && value.range.data_type().is_integer() + { + Ok(value.sort_properties) + } else { + Ok(SortProperties::Unordered) + } + } + fn documentation(&self) -> Option<&Documentation> { self.doc() } @@ -719,6 +740,16 @@ impl ScalarUDFImpl for ToTimestampMicrosFunc { } } + fn output_ordering(&self, input: &[ExprProperties]) -> Result { + if let [value] = input + && value.range.data_type().is_integer() + { + Ok(value.sort_properties) + } else { + Ok(SortProperties::Unordered) + } + } + fn documentation(&self) -> Option<&Documentation> { self.doc() } @@ -788,6 +819,16 @@ impl ScalarUDFImpl for ToTimestampNanosFunc { } } + fn output_ordering(&self, input: &[ExprProperties]) -> Result { + if let [value] = input + && value.range.data_type().is_integer() + { + Ok(value.sort_properties) + } else { + Ok(SortProperties::Unordered) + } + } + fn documentation(&self) -> Option<&Documentation> { self.doc() } @@ -852,6 +893,66 @@ mod tests { use super::*; + #[test] + fn integer_timestamp_ordering() -> Result<()> { + use arrow::compute::SortOptions; + use datafusion_expr::interval_arithmetic::Interval; + + let config = ConfigOptions::default(); + let functions: [Box; 4] = [ + Box::new(ToTimestampSecondsFunc::new_with_config(&config)), + Box::new(ToTimestampMillisFunc::new_with_config(&config)), + Box::new(ToTimestampMicrosFunc::new_with_config(&config)), + Box::new(ToTimestampNanosFunc::new_with_config(&config)), + ]; + for function in functions { + for descending in [false, true] { + for nulls_first in [false, true] { + let ordering = SortProperties::Ordered(SortOptions { + descending, + nulls_first, + }); + for (data_type, supported) in [ + (Int8, true), + (Int16, true), + (Int32, true), + (Int64, true), + (UInt8, true), + (UInt16, true), + (UInt32, true), + (UInt64, true), + (Float64, false), + (Decimal128(20, 0), false), + (Date32, false), + (Date64, false), + (Utf8, false), + (Null, false), + (Timestamp(Microsecond, None), false), + ] { + let input = ExprProperties::new_unknown() + .with_order(ordering) + .with_range(Interval::make_unbounded(&data_type)?); + let expected = if supported { + ordering + } else { + SortProperties::Unordered + }; + assert_eq!( + function.output_ordering(std::slice::from_ref(&input))?, + expected + ); + assert_eq!( + function.output_ordering(&[input.clone(), input])?, + SortProperties::Unordered + ); + } + } + } + assert_eq!(function.output_ordering(&[])?, SortProperties::Unordered); + } + Ok(()) + } + fn to_timestamp(args: &[ColumnarValue]) -> Result { let timezone: Option> = Some("UTC".into()); to_timestamp_impl::( diff --git a/datafusion/functions/src/datetime/to_unixtime.rs b/datafusion/functions/src/datetime/to_unixtime.rs index 5b9734c05d7be..17b01e46be1fb 100644 --- a/datafusion/functions/src/datetime/to_unixtime.rs +++ b/datafusion/functions/src/datetime/to_unixtime.rs @@ -19,6 +19,7 @@ use super::to_timestamp::ToTimestampSecondsFunc; use crate::datetime::common::*; use arrow::datatypes::{DataType, TimeUnit}; use datafusion_common::{Result, exec_err}; +use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use datafusion_expr::{ ColumnarValue, Documentation, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, @@ -143,7 +144,135 @@ impl ScalarUDFImpl for ToUnixtimeFunc { } } + fn output_ordering(&self, input: &[ExprProperties]) -> Result { + if let [value] = input + && (value.range.data_type().is_integer() + || matches!( + value.range.data_type(), + DataType::Date32 | DataType::Date64 | DataType::Timestamp(_, _) + )) + { + // Conversions preserve epoch order and report overflow instead of adding NULLs. + Ok(value.sort_properties) + } else { + Ok(SortProperties::Unordered) + } + } + fn documentation(&self) -> Option<&Documentation> { self.doc() } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow::array::{Array, Int64Array}; + use arrow::compute::{SortOptions, sort}; + use arrow::datatypes::DataType::*; + use datafusion_common::{config::ConfigOptions, datatype::DataTypeExt}; + use datafusion_expr::interval_arithmetic::Interval; + + use super::*; + + #[test] + fn ordering_matches_runtime_conversion() -> Result<()> { + let function = ToUnixtimeFunc::new(); + let mut cases = vec![ + (Int8, true), + (Int16, true), + (Int32, true), + (Int64, true), + (UInt8, true), + (UInt16, true), + (UInt32, true), + (UInt64, true), + (Date32, true), + (Date64, true), + (Float64, false), + (Decimal128(20, 0), false), + (Utf8, false), + (Null, false), + ]; + for unit in [ + TimeUnit::Second, + TimeUnit::Millisecond, + TimeUnit::Microsecond, + TimeUnit::Nanosecond, + ] { + for timezone in [ + None, + Some("UTC".into()), + Some("+05:30".into()), + Some("America/New_York".into()), + ] { + cases.push((Timestamp(unit, timezone), true)); + } + } + for (data_type, supported) in cases { + for (descending, nulls_first) in + [(false, false), (false, true), (true, false), (true, true)] + { + let options = SortOptions { + descending, + nulls_first, + }; + let ordering = SortProperties::Ordered(options); + let input = ExprProperties::new_unknown() + .with_order(ordering) + .with_range(Interval::make_unbounded(&data_type)?); + assert_eq!( + function.output_ordering(&[input])?, + if supported { + ordering + } else { + SortProperties::Unordered + } + ); + if !supported || data_type.is_integer() { + continue; + } + + let (min, max) = if data_type == Date32 { + (i32::MIN as i64, i32::MAX as i64) + } else { + (i64::MIN, i64::MAX) + }; + let values = Int64Array::from(vec![ + Some(min), + Some(-1), + Some(0), + Some(1), + Some(max), + None, + ]); + let argument = ColumnarValue::Array(sort(&values, Some(options))?) + .cast_to(&data_type, None)?; + let result = function + .invoke_with_args(ScalarFunctionArgs { + args: vec![argument], + arg_fields: vec![data_type.clone().into_nullable_field_ref()], + number_rows: values.len(), + return_field: Int64.into_nullable_field_ref(), + config_options: Arc::new(ConfigOptions::default()), + })? + .to_array(values.len())?; + assert_eq!( + result.as_ref(), + sort(result.as_ref(), Some(options))?.as_ref() + ); + assert_eq!(result.null_count(), values.null_count()); + } + } + let input = ExprProperties::new_unknown() + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_range(Interval::make_unbounded(&Int64)?); + assert_eq!( + function.output_ordering(&[input.clone(), input])?, + SortProperties::Unordered + ); + assert_eq!(function.output_ordering(&[])?, SortProperties::Unordered); + Ok(()) + } +} diff --git a/datafusion/sqllogictest/test_files/datetime/timestamps.slt b/datafusion/sqllogictest/test_files/datetime/timestamps.slt index 71118ddcabb73..0fe0e695387b2 100644 --- a/datafusion/sqllogictest/test_files/datetime/timestamps.slt +++ b/datafusion/sqllogictest/test_files/datetime/timestamps.slt @@ -6053,3 +6053,142 @@ query P SELECT date_bin(NULL, TIMESTAMP '2023-01-01 12:30:00', TIMESTAMP '2023-01-01 12:00:00') ---- NULL + + +# Timestamp conversion ordering. +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.explain.physical_plan_only = true; + +query I +COPY ( + SELECT column1::BIGINT AS micros, column2::INT AS id, + arrow_cast(column1, 'Timestamp(Microsecond, None)') AS ts + FROM (VALUES (-1, 1), (0, 0), (1000000, 2), (NULL, 3)) + ORDER BY micros ASC NULLS LAST, id +) TO 'test_files/scratch/timestamps/conversion_asc.parquet' STORED AS PARQUET; +---- +4 + +statement ok +CREATE EXTERNAL TABLE ordered_timestamps STORED AS PARQUET +WITH ORDER (micros ASC NULLS LAST, id ASC) +WITH ORDER (ts ASC NULLS LAST, id ASC) +LOCATION 'test_files/scratch/timestamps/conversion_asc.parquet'; + +# The conversion chain preserves order: no SortExec is needed. +query TT +EXPLAIN SELECT to_unixtime(date_bin(interval '1 second', to_timestamp_micros(micros), timestamp '1970-01-01')) AS bucket +FROM ordered_timestamps ORDER BY bucket ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_asc.parquet]]}, projection=[to_unixtime(date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 1000000000 }, to_timestamp_micros(micros@0), 0)) as bucket], file_type=parquet + +query I +SELECT to_unixtime(date_bin(interval '1 second', to_timestamp_micros(micros), timestamp '1970-01-01')) AS bucket +FROM ordered_timestamps ORDER BY bucket ASC NULLS LAST; +---- +-1 +0 +1 +NULL + +# Truncation creates ties, so sorting by seconds then id still needs SortExec. +query TT +EXPLAIN SELECT to_unixtime(ts) AS seconds, id FROM ordered_timestamps ORDER BY seconds ASC NULLS LAST, id; +---- +physical_plan +01)SortExec: expr=[seconds@0 ASC NULLS LAST, id@1 ASC NULLS LAST], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_asc.parquet]]}, projection=[to_unixtime(ts@2) as seconds, id], file_type=parquet + +query II +SELECT to_unixtime(ts) AS seconds, id FROM ordered_timestamps ORDER BY seconds ASC NULLS LAST, id; +---- +0 0 +0 1 +1 2 +NULL 3 + +# INT32, UInt64, Date32 and Date64 conversions also preserve order. +query I +COPY ( + SELECT column1::INT AS i, arrow_cast(column2, 'UInt64') AS u, column3::DATE AS d + FROM (VALUES (-1, 0, '1969-12-31'), (0, 1, '1970-01-01'), + (1, 9223372036854775807, '1970-01-02'), (NULL, NULL, NULL)) + ORDER BY i ASC NULLS LAST +) TO 'test_files/scratch/timestamps/conversion_integers.parquet' STORED AS PARQUET; +---- +4 + +statement ok +CREATE EXTERNAL TABLE ordered_conversions STORED AS PARQUET +WITH ORDER (i ASC NULLS LAST) +WITH ORDER (u ASC NULLS LAST) +WITH ORDER (d ASC NULLS LAST) +LOCATION 'test_files/scratch/timestamps/conversion_integers.parquet'; + +query TT +EXPLAIN SELECT to_timestamp_micros(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_timestamp_micros(i@0) as t], file_type=parquet + +query TT +EXPLAIN SELECT to_timestamp_seconds(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_timestamp_seconds(i@0) as t], file_type=parquet + +query TT +EXPLAIN SELECT to_timestamp_millis(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_timestamp_millis(i@0) as t], file_type=parquet + +query TT +EXPLAIN SELECT to_timestamp_nanos(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_timestamp_nanos(i@0) as t], file_type=parquet + +query PPPP +SELECT to_timestamp_seconds(i) AS seconds, to_timestamp_millis(i) AS millis, + to_timestamp_micros(i) AS micros, to_timestamp_nanos(i) AS nanos +FROM ordered_conversions ORDER BY seconds ASC NULLS LAST; +---- +1969-12-31T23:59:59 1969-12-31T23:59:59.999 1969-12-31T23:59:59.999999 1969-12-31T23:59:59.999999999 +1970-01-01T00:00:00 1970-01-01T00:00:00 1970-01-01T00:00:00 1970-01-01T00:00:00 +1970-01-01T00:00:01 1970-01-01T00:00:00.001 1970-01-01T00:00:00.000001 1970-01-01T00:00:00.000000001 +NULL NULL NULL NULL + +query TT +EXPLAIN SELECT to_unixtime(u) AS seconds FROM ordered_conversions ORDER BY seconds ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_unixtime(u@1) as seconds], file_type=parquet + +query TT +EXPLAIN SELECT to_unixtime(d) AS seconds FROM ordered_conversions ORDER BY seconds ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_unixtime(d@2) as seconds], file_type=parquet + +query TT +EXPLAIN SELECT to_unixtime(arrow_cast(d, 'Date64')) AS seconds FROM ordered_conversions ORDER BY seconds ASC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/timestamps/conversion_integers.parquet]]}, projection=[to_unixtime(CAST(d@2 AS Date64)) as seconds], file_type=parquet + +# Check UInt64's largest valid value using raw microseconds, and both date results. +query IIII +SELECT arrow_cast(to_timestamp_micros(u), 'Int64') AS micros, to_unixtime(u) AS seconds, + to_unixtime(d) AS date32_seconds, to_unixtime(arrow_cast(d, 'Date64')) AS date64_seconds +FROM ordered_conversions ORDER BY u ASC NULLS LAST; +---- +0 0 -86400 -86400 +1 1 0 0 +9223372036854775807 9223372036854775807 86400 86400 +NULL NULL NULL NULL + +query error Can't cast value 9223372036854775808 to type Int64 +SELECT to_timestamp_micros(arrow_cast(9223372036854775808, 'UInt64')); + +statement ok +RESET datafusion.explain.physical_plan_only; + +statement ok +SET datafusion.execution.target_partitions = 4;