From 1f5219198c513337a0420e2b59a8a6dbf3107f3c Mon Sep 17 00:00:00 2001 From: Huaijin Date: Wed, 7 Oct 2026 18:22:35 +0800 Subject: [PATCH 1/2] feat: preserve ordering through to_timestamp_micros and to_unixtime --- .../functions/src/datetime/to_timestamp.rs | 101 +++++++++++ .../functions/src/datetime/to_unixtime.rs | 163 ++++++++++++++++++ .../test_files/datetime/timestamps.slt | 126 ++++++++++++++ 3 files changed, 390 insertions(+) diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs index 6720152cbd3a7..049c667a727ba 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, @@ -719,6 +720,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() } @@ -852,6 +863,96 @@ mod tests { use super::*; + #[test] + fn micros_integer_ordering_matches_runtime_conversion() -> Result<()> { + use arrow::compute::{SortOptions, sort}; + use datafusion_expr::interval_arithmetic::Interval; + + let function = ToTimestampMicrosFunc::new_with_config(&ConfigOptions::default()); + for descending in [false, true] { + for nulls_first in [false, true] { + let ordering = SortProperties::Ordered(SortOptions { + descending, + nulls_first, + }); + for data_type in [ + Int8, + Int16, + Int32, + Int64, + UInt8, + UInt16, + UInt32, + UInt64, + Float64, + Decimal128(20, 0), + Date32, + Date64, + Utf8, + Null, + Timestamp(Microsecond, None), + ] { + let input = ExprProperties::new_unknown() + .with_order(ordering) + .with_range(Interval::make_unbounded(&data_type)?); + let expected = if data_type.is_integer() { + ordering + } else { + SortProperties::Unordered + }; + assert_eq!( + function.output_ordering(std::slice::from_ref(&input))?, + expected + ); + if data_type.is_integer() { + let values = if data_type.is_signed_integer() { + Int64Array::from(vec![Some(-1), Some(0), Some(1), None]) + } else { + Int64Array::from(vec![Some(0), Some(1), Some(2), None]) + }; + let options = SortOptions { + descending, + nulls_first, + }; + let sorted = sort(&values, Some(options))?; + let argument = ColumnarValue::Array(Arc::clone(&sorted)) + .cast_to(&data_type, None)?; + let result = function.invoke_with_args(ScalarFunctionArgs { + args: vec![argument], + arg_fields: vec![Arc::new(Field::new( + "micros", + data_type.clone(), + true, + ))], + number_rows: values.len(), + return_field: Arc::new(Field::new( + "ts", + Timestamp(Microsecond, None), + true, + )), + config_options: Arc::new(ConfigOptions::default()), + })?; + let ColumnarValue::Array(result) = + result.cast_to(&Int64, None)? + else { + panic!("expected array") + }; + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + sorted.as_any().downcast_ref::().unwrap() + ); + } + 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..b6c236ab90b76 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,169 @@ 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::Field; + use datafusion_common::config::ConfigOptions; + use datafusion_expr::interval_arithmetic::Interval; + + use super::*; + + #[test] + fn supported_ordering_matches_runtime_conversion() -> Result<()> { + let function = ToUnixtimeFunc::new(); + let values = Int64Array::from(vec![ + Some(i64::MIN), + Some(-1000001), + Some(-1), + Some(0), + Some(1), + Some(1000001), + Some(i64::MAX), + None, + ]); + let mut inputs = vec![ + (DataType::Int8, i8::MIN as i64, i8::MAX as i64), + (DataType::Int16, i16::MIN as i64, i16::MAX as i64), + (DataType::Int32, i32::MIN as i64, i32::MAX as i64), + (DataType::Int64, i64::MIN, i64::MAX), + (DataType::UInt8, 0, u8::MAX as i64), + (DataType::UInt16, 0, u16::MAX as i64), + (DataType::UInt32, 0, u32::MAX as i64), + (DataType::UInt64, 0, i64::MAX), + (DataType::Date32, i32::MIN as i64, i32::MAX as i64), + (DataType::Date64, i64::MIN, i64::MAX), + ] + .into_iter() + .map(|(data_type, min, max)| { + ( + data_type, + Int64Array::from(vec![Some(min), Some(0), Some(max), None]), + ) + }) + .collect::>(); + inputs.push((DataType::Date64, values.clone())); + 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()), + ] { + inputs.push((DataType::Timestamp(unit, timezone), values.clone())); + } + } + for (data_type, values) in inputs { + for descending in [false, true] { + for nulls_first in [false, 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(std::slice::from_ref(&input))?, + ordering + ); + assert!( + !function + .strictly_order_preserving(std::slice::from_ref(&input))? + ); + assert!(!function.preserves_lex_ordering(&[input])?); + + let sorted = sort(&values, Some(options))?; + let argument = + ColumnarValue::Array(sorted).cast_to(&data_type, None)?; + let result = function.invoke_with_args(ScalarFunctionArgs { + args: vec![argument], + arg_fields: vec![Arc::new(Field::new( + "ts", + data_type.clone(), + true, + ))], + number_rows: values.len(), + return_field: Arc::new(Field::new( + "seconds", + DataType::Int64, + true, + )), + config_options: Arc::new(ConfigOptions::default()), + })?; + let ColumnarValue::Array(result) = result else { + panic!("expected array") + }; + let expected = sort(result.as_ref(), Some(options))?; + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + expected.as_any().downcast_ref::().unwrap() + ); + assert_eq!(result.null_count(), values.null_count()); + } + } + } + Ok(()) + } + + #[test] + fn ordering_rejects_unsupported_types_and_multiple_arguments() -> Result<()> { + let function = ToUnixtimeFunc::new(); + for data_type in [ + DataType::Float64, + DataType::Decimal128(20, 0), + DataType::Utf8, + DataType::Null, + ] { + let input = ExprProperties::new_unknown() + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_range(Interval::make_unbounded(&data_type)?); + assert_eq!( + function.output_ordering(&[input])?, + SortProperties::Unordered + ); + } + let timestamp = ExprProperties::new_unknown() + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_range(Interval::make_unbounded(&DataType::Timestamp( + TimeUnit::Microsecond, + None, + ))?); + assert_eq!( + function.output_ordering(&[timestamp.clone(), timestamp])?, + 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..bf04141a14be0 100644 --- a/datafusion/sqllogictest/test_files/datetime/timestamps.slt +++ b/datafusion/sqllogictest/test_files/datetime/timestamps.slt @@ -6053,3 +6053,129 @@ 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 (-1000001, 0), (-1, 3), (0, 2), (1, 1), (1000000, 4), (NULL, 5)) + ORDER BY micros ASC NULLS LAST, id +) TO 'test_files/scratch/timestamps/conversion_asc.parquet' STORED AS PARQUET; +---- +6 + +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; +---- +-2 +-1 +0 +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; +---- +-1 0 +0 1 +0 2 +0 3 +1 4 +NULL 5 + +# 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 P +SELECT to_timestamp_micros(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +---- +1969-12-31T23:59:59.999999 +1970-01-01T00:00:00 +1970-01-01T00:00:00.000001 +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; From 1c88c3da3a35a1e7d56ea11539f79fd9b97cb10f Mon Sep 17 00:00:00 2001 From: Huaijin Date: Wed, 7 Oct 2026 18:55:19 +0800 Subject: [PATCH 2/2] feat: preserve ordering through all timestamp unit conversions --- .../functions/src/datetime/to_timestamp.rs | 152 +++++++-------- .../functions/src/datetime/to_unixtime.rs | 176 +++++++----------- .../test_files/datetime/timestamps.slt | 43 +++-- 3 files changed, 175 insertions(+), 196 deletions(-) diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs index 049c667a727ba..6e6ee1ba183e7 100644 --- a/datafusion/functions/src/datetime/to_timestamp.rs +++ b/datafusion/functions/src/datetime/to_timestamp.rs @@ -582,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() } @@ -651,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() } @@ -799,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() } @@ -864,92 +894,62 @@ mod tests { use super::*; #[test] - fn micros_integer_ordering_matches_runtime_conversion() -> Result<()> { - use arrow::compute::{SortOptions, sort}; + fn integer_timestamp_ordering() -> Result<()> { + use arrow::compute::SortOptions; use datafusion_expr::interval_arithmetic::Interval; - let function = ToTimestampMicrosFunc::new_with_config(&ConfigOptions::default()); - for descending in [false, true] { - for nulls_first in [false, true] { - let ordering = SortProperties::Ordered(SortOptions { - descending, - nulls_first, - }); - for data_type in [ - Int8, - Int16, - Int32, - Int64, - UInt8, - UInt16, - UInt32, - UInt64, - Float64, - Decimal128(20, 0), - Date32, - Date64, - Utf8, - Null, - Timestamp(Microsecond, None), - ] { - let input = ExprProperties::new_unknown() - .with_order(ordering) - .with_range(Interval::make_unbounded(&data_type)?); - let expected = if data_type.is_integer() { - ordering - } else { - SortProperties::Unordered - }; - assert_eq!( - function.output_ordering(std::slice::from_ref(&input))?, - expected - ); - if data_type.is_integer() { - let values = if data_type.is_signed_integer() { - Int64Array::from(vec![Some(-1), Some(0), Some(1), None]) + 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 { - Int64Array::from(vec![Some(0), Some(1), Some(2), None]) - }; - let options = SortOptions { - descending, - nulls_first, - }; - let sorted = sort(&values, Some(options))?; - let argument = ColumnarValue::Array(Arc::clone(&sorted)) - .cast_to(&data_type, None)?; - let result = function.invoke_with_args(ScalarFunctionArgs { - args: vec![argument], - arg_fields: vec![Arc::new(Field::new( - "micros", - data_type.clone(), - true, - ))], - number_rows: values.len(), - return_field: Arc::new(Field::new( - "ts", - Timestamp(Microsecond, None), - true, - )), - config_options: Arc::new(ConfigOptions::default()), - })?; - let ColumnarValue::Array(result) = - result.cast_to(&Int64, None)? - else { - panic!("expected array") + SortProperties::Unordered }; assert_eq!( - result.as_any().downcast_ref::().unwrap(), - sorted.as_any().downcast_ref::().unwrap() + function.output_ordering(std::slice::from_ref(&input))?, + expected + ); + assert_eq!( + function.output_ordering(&[input.clone(), input])?, + SortProperties::Unordered ); } - assert_eq!( - function.output_ordering(&[input.clone(), input])?, - SortProperties::Unordered - ); } } + assert_eq!(function.output_ordering(&[])?, SortProperties::Unordered); } - assert_eq!(function.output_ordering(&[])?, SortProperties::Unordered); Ok(()) } diff --git a/datafusion/functions/src/datetime/to_unixtime.rs b/datafusion/functions/src/datetime/to_unixtime.rs index b6c236ab90b76..17b01e46be1fb 100644 --- a/datafusion/functions/src/datetime/to_unixtime.rs +++ b/datafusion/functions/src/datetime/to_unixtime.rs @@ -170,46 +170,31 @@ mod tests { use arrow::array::{Array, Int64Array}; use arrow::compute::{SortOptions, sort}; - use arrow::datatypes::Field; - use datafusion_common::config::ConfigOptions; + use arrow::datatypes::DataType::*; + use datafusion_common::{config::ConfigOptions, datatype::DataTypeExt}; use datafusion_expr::interval_arithmetic::Interval; use super::*; #[test] - fn supported_ordering_matches_runtime_conversion() -> Result<()> { + fn ordering_matches_runtime_conversion() -> Result<()> { let function = ToUnixtimeFunc::new(); - let values = Int64Array::from(vec![ - Some(i64::MIN), - Some(-1000001), - Some(-1), - Some(0), - Some(1), - Some(1000001), - Some(i64::MAX), - None, - ]); - let mut inputs = vec![ - (DataType::Int8, i8::MIN as i64, i8::MAX as i64), - (DataType::Int16, i16::MIN as i64, i16::MAX as i64), - (DataType::Int32, i32::MIN as i64, i32::MAX as i64), - (DataType::Int64, i64::MIN, i64::MAX), - (DataType::UInt8, 0, u8::MAX as i64), - (DataType::UInt16, 0, u16::MAX as i64), - (DataType::UInt32, 0, u32::MAX as i64), - (DataType::UInt64, 0, i64::MAX), - (DataType::Date32, i32::MIN as i64, i32::MAX as i64), - (DataType::Date64, i64::MIN, i64::MAX), - ] - .into_iter() - .map(|(data_type, min, max)| { - ( - data_type, - Int64Array::from(vec![Some(min), Some(0), Some(max), None]), - ) - }) - .collect::>(); - inputs.push((DataType::Date64, values.clone())); + 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, @@ -222,88 +207,69 @@ mod tests { Some("+05:30".into()), Some("America/New_York".into()), ] { - inputs.push((DataType::Timestamp(unit, timezone), values.clone())); + cases.push((Timestamp(unit, timezone), true)); } } - for (data_type, values) in inputs { - for descending in [false, true] { - for nulls_first in [false, 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(std::slice::from_ref(&input))?, + 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 - ); - assert!( - !function - .strictly_order_preserving(std::slice::from_ref(&input))? - ); - assert!(!function.preserves_lex_ordering(&[input])?); + } else { + SortProperties::Unordered + } + ); + if !supported || data_type.is_integer() { + continue; + } - let sorted = sort(&values, Some(options))?; - let argument = - ColumnarValue::Array(sorted).cast_to(&data_type, None)?; - let result = function.invoke_with_args(ScalarFunctionArgs { + 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![Arc::new(Field::new( - "ts", - data_type.clone(), - true, - ))], + arg_fields: vec![data_type.clone().into_nullable_field_ref()], number_rows: values.len(), - return_field: Arc::new(Field::new( - "seconds", - DataType::Int64, - true, - )), + return_field: Int64.into_nullable_field_ref(), config_options: Arc::new(ConfigOptions::default()), - })?; - let ColumnarValue::Array(result) = result else { - panic!("expected array") - }; - let expected = sort(result.as_ref(), Some(options))?; - assert_eq!( - result.as_any().downcast_ref::().unwrap(), - expected.as_any().downcast_ref::().unwrap() - ); - assert_eq!(result.null_count(), values.null_count()); - } + })? + .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()); } } - Ok(()) - } - - #[test] - fn ordering_rejects_unsupported_types_and_multiple_arguments() -> Result<()> { - let function = ToUnixtimeFunc::new(); - for data_type in [ - DataType::Float64, - DataType::Decimal128(20, 0), - DataType::Utf8, - DataType::Null, - ] { - let input = ExprProperties::new_unknown() - .with_order(SortProperties::Ordered(SortOptions::default())) - .with_range(Interval::make_unbounded(&data_type)?); - assert_eq!( - function.output_ordering(&[input])?, - SortProperties::Unordered - ); - } - let timestamp = ExprProperties::new_unknown() + let input = ExprProperties::new_unknown() .with_order(SortProperties::Ordered(SortOptions::default())) - .with_range(Interval::make_unbounded(&DataType::Timestamp( - TimeUnit::Microsecond, - None, - ))?); + .with_range(Interval::make_unbounded(&Int64)?); assert_eq!( - function.output_ordering(&[timestamp.clone(), timestamp])?, + function.output_ordering(&[input.clone(), input])?, SortProperties::Unordered ); assert_eq!(function.output_ordering(&[])?, SortProperties::Unordered); diff --git a/datafusion/sqllogictest/test_files/datetime/timestamps.slt b/datafusion/sqllogictest/test_files/datetime/timestamps.slt index bf04141a14be0..0fe0e695387b2 100644 --- a/datafusion/sqllogictest/test_files/datetime/timestamps.slt +++ b/datafusion/sqllogictest/test_files/datetime/timestamps.slt @@ -6066,11 +6066,11 @@ query I COPY ( SELECT column1::BIGINT AS micros, column2::INT AS id, arrow_cast(column1, 'Timestamp(Microsecond, None)') AS ts - FROM (VALUES (-1000001, 0), (-1, 3), (0, 2), (1, 1), (1000000, 4), (NULL, 5)) + 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; ---- -6 +4 statement ok CREATE EXTERNAL TABLE ordered_timestamps STORED AS PARQUET @@ -6089,10 +6089,8 @@ 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; ---- --2 -1 0 -0 1 NULL @@ -6107,12 +6105,10 @@ physical_plan query II SELECT to_unixtime(ts) AS seconds, id FROM ordered_timestamps ORDER BY seconds ASC NULLS LAST, id; ---- --1 0 +0 0 0 1 -0 2 -0 3 -1 4 -NULL 5 +1 2 +NULL 3 # INT32, UInt64, Date32 and Date64 conversions also preserve order. query I @@ -6137,13 +6133,30 @@ EXPLAIN SELECT to_timestamp_micros(i) AS t FROM ordered_conversions ORDER BY t A ---- 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 P -SELECT to_timestamp_micros(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; +query TT +EXPLAIN SELECT to_timestamp_seconds(i) AS t FROM ordered_conversions ORDER BY t ASC NULLS LAST; ---- -1969-12-31T23:59:59.999999 -1970-01-01T00:00:00 -1970-01-01T00:00:00.000001 -NULL +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;