diff --git a/src/query/src/optimizer/windowed_sort.rs b/src/query/src/optimizer/windowed_sort.rs index 62c17c40b7a..d1ec687c6cf 100644 --- a/src/query/src/optimizer/windowed_sort.rs +++ b/src/query/src/optimizer/windowed_sort.rs @@ -222,7 +222,8 @@ fn is_time_index_expr( } if let Some(cast_expr) = expr.downcast_ref::() { - return if matches!(cast_expr.cast_type(), DataType::Timestamp(_, _)) { + let source_type = cast_expr.expr().data_type(plan.schema().as_ref())?; + return if preserves_timestamp_values(cast_expr.cast_type(), &source_type) { is_time_index_expr(plan, cast_expr.expr()) } else { Ok(false) @@ -230,10 +231,22 @@ fn is_time_index_expr( } if let Some(scalar_function_expr) = expr.downcast_ref::() { - return if is_supported_time_index_wrapper(scalar_function_expr) - && scalar_function_expr.args().len() == 1 + if !is_supported_time_index_wrapper(scalar_function_expr) + || scalar_function_expr.args().len() != 1 { - is_time_index_expr(plan, &scalar_function_expr.args()[0]) + return Ok(false); + } + + // Wrappers such as `to_timestamp_seconds` cast their argument to another time unit + // inside the function. That cast does not show up as a `CastExpr` in the plan, but it + // changes the values of the sort column exactly like a cast in the plan does, so the + // wrapper is only still the time index if its result keeps both the unit and the values + // of its argument. + let arg = &scalar_function_expr.args()[0]; + let return_type = scalar_function_expr.data_type(plan.schema().as_ref())?; + let arg_type = arg.data_type(plan.schema().as_ref())?; + return if preserves_timestamp_values(&return_type, &arg_type) { + is_time_index_expr(plan, arg) } else { Ok(false) }; @@ -242,6 +255,24 @@ fn is_time_index_expr( Ok(false) } +/// Returns true if a cast, or a function wrapper, that turns `source` into `target` keeps both the +/// time unit and the underlying values of a timestamp. +/// +/// A cast (or a hidden cast inside a function wrapper) that changes the time unit changes the +/// values of the sort column, so the working ranges derived from the time index, which keep the +/// unit of the time index, no longer apply. The optimizer conservatively rejects every cast that +/// adds a timezone to a naive timestamp, including UTC, even though UTC has a zero offset. Only a +/// target that keeps both the unit and the values of its source can still be the time index. +fn preserves_timestamp_values(target: &DataType, source: &DataType) -> bool { + matches!( + (target, source), + ( + DataType::Timestamp(target_unit, target_tz), + DataType::Timestamp(source_unit, source_tz) + ) if target_unit == source_unit && (source_tz.is_some() || target_tz.is_none()) + ) +} + fn is_time_index_column( plan: &Arc, column_expr: &PhysicalColumn, @@ -344,7 +375,7 @@ mod test { use datafusion::config::ConfigOptions; use datafusion::physical_plan::filter::FilterExecBuilder; use datafusion_common::ScalarValue; - use datafusion_functions::datetime::to_timestamp_millis; + use datafusion_functions::datetime::{to_timestamp, to_timestamp_millis, to_timestamp_seconds}; use datafusion_physical_expr::expressions::{CastExpr, Literal}; use datatypes::data_type::ConcreteDataType; use datatypes::schema::{ColumnSchema, Schema}; @@ -416,37 +447,92 @@ mod test { } #[test] - fn test_is_time_index_expr_tracks_wrapped_aliases_through_projection() { - let scan = new_region_scan(); + fn test_is_time_index_expr_tracks_value_preserving_wrapper_through_projection() { + let scan = new_region_scan_with_unit(TimeUnit::Millisecond); let config = Arc::new(ConfigOptions::default()); - let return_field = Arc::new(Field::new( - "ts", - DataType::Timestamp(TimeUnit::Millisecond, None), - false, - )); + let wrapper = ScalarFunctionExpr::try_new( + to_timestamp_millis(config.as_ref()), + vec![Arc::new(PhysicalColumn::new("ts", 1))], + scan.schema().as_ref(), + config, + ) + .unwrap(); + // `to_timestamp_millis` keeps both the unit and the values of a millisecond timestamp. + assert_eq!( + wrapper.data_type(scan.schema().as_ref()).unwrap(), + DataType::Timestamp(TimeUnit::Millisecond, None) + ); let projection = Arc::new( ProjectionExec::try_new( - vec![( - Arc::new(ScalarFunctionExpr::new( - "to_timestamp_millis", - to_timestamp_millis(config.as_ref()), - vec![Arc::new(PhysicalColumn::new("ts", 1))], - return_field, - config, - )) as Arc, - "ts".to_string(), - )], + vec![(Arc::new(wrapper) as Arc, "ts".to_string())], scan, ) .unwrap(), ) as Arc; let expr = Arc::new(PhysicalColumn::new("ts", 0)) as Arc; + // So the wrapper is still the time index. assert!(is_time_index_expr(&projection, &expr).unwrap()); } #[test] - fn test_is_time_index_expr_tracks_cast_aliases_through_projection() { + fn test_is_time_index_expr_rejects_unit_changing_wrapper() { + let config = Arc::new(ConfigOptions::default()); + + // `to_timestamp_millis` casts its argument to milliseconds, which truncates the values of + // a nanosecond time index, so the wrapper is no longer the time index. + let nanosecond_scan = new_region_scan(); + let truncates_to_millis = Arc::new( + ScalarFunctionExpr::try_new( + to_timestamp_millis(config.as_ref()), + vec![Arc::new(PhysicalColumn::new("ts", 1))], + nanosecond_scan.schema().as_ref(), + config.clone(), + ) + .unwrap(), + ) as Arc; + assert_eq!( + truncates_to_millis + .data_type(nanosecond_scan.schema().as_ref()) + .unwrap(), + DataType::Timestamp(TimeUnit::Millisecond, None) + ); + assert!(!is_time_index_expr(&nanosecond_scan, &truncates_to_millis).unwrap()); + + // Same for a millisecond time index wrapped in `to_timestamp_seconds`. + let millisecond_scan = new_region_scan_with_unit(TimeUnit::Millisecond); + let truncates_to_seconds = Arc::new( + ScalarFunctionExpr::try_new( + to_timestamp_seconds(config.as_ref()), + vec![Arc::new(PhysicalColumn::new("ts", 1))], + millisecond_scan.schema().as_ref(), + config.clone(), + ) + .unwrap(), + ) as Arc; + assert_eq!( + truncates_to_seconds + .data_type(millisecond_scan.schema().as_ref()) + .unwrap(), + DataType::Timestamp(TimeUnit::Second, None) + ); + assert!(!is_time_index_expr(&millisecond_scan, &truncates_to_seconds).unwrap()); + + // A supported wrapper whose argument is not a timestamp is never the time index. + let from_integer = Arc::new( + ScalarFunctionExpr::try_new( + to_timestamp_millis(config.as_ref()), + vec![Arc::new(Literal::new(ScalarValue::Int64(Some(0))))], + nanosecond_scan.schema().as_ref(), + config, + ) + .unwrap(), + ) as Arc; + assert!(!is_time_index_expr(&nanosecond_scan, &from_integer).unwrap()); + } + + #[test] + fn test_is_time_index_expr_rejects_unit_changing_cast_through_projection() { let scan = new_region_scan(); let projection = Arc::new( ProjectionExec::try_new( @@ -464,7 +550,116 @@ mod test { ) as Arc; let expr = Arc::new(PhysicalColumn::new("ts_ms", 0)) as Arc; - assert!(is_time_index_expr(&projection, &expr).unwrap()); + // The cast changes the time unit (ns → ms), so it is no longer the time index. + assert!(!is_time_index_expr(&projection, &expr).unwrap()); + } + + #[test] + fn test_is_time_index_expr_rejects_timezone_adding_cast() { + let scan = new_region_scan(); + + // The optimizer conservatively rejects casts adding any timezone to a naive timestamp. + let adds_timezone = Arc::new(CastExpr::new( + Arc::new(PhysicalColumn::new("ts", 1)), + DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+07:00"))), + None, + )) as Arc; + assert!(!is_time_index_expr(&scan, &adds_timezone).unwrap()); + + // UTC has a zero offset, but naive-to-aware timestamp casts are all rejected. + let adds_utc_timezone = Arc::new(CastExpr::new( + Arc::new(PhysicalColumn::new("ts", 1)), + DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("UTC"))), + None, + )) as Arc; + assert!(!is_time_index_expr(&scan, &adds_utc_timezone).unwrap()); + + // `to_timestamp` returns an aware timestamp in the session time zone for a naive + // timestamp argument. That re-interprets the underlying values in the same way as a + // timezone-adding cast, so the wrapper is rejected as well. + let mut config = ConfigOptions::default(); + config.execution.time_zone = Some("+08:00".to_string()); + let config = Arc::new(config); + let aware_source = ScalarFunctionExpr::try_new( + to_timestamp(config.as_ref()), + vec![Arc::new(PhysicalColumn::new("ts", 1))], + scan.schema().as_ref(), + config.clone(), + ) + .unwrap(); + let projection = Arc::new( + ProjectionExec::try_new( + vec![( + Arc::new(aware_source) as Arc, + "ts_aware".to_string(), + )], + scan, + ) + .unwrap(), + ) as Arc; + assert_eq!( + projection.schema().field(0).data_type(), + &DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+08:00"))) + ); + let expr = Arc::new(PhysicalColumn::new("ts_aware", 0)) as Arc; + + assert!(!is_time_index_expr(&projection, &expr).unwrap()); + + // A cast between two aware timestamps of the same unit keeps the underlying values, see + // `test_preserves_timestamp_values`, but it is still rejected here as its input is not + // the time index. + let aware_cast = Arc::new(CastExpr::new( + Arc::new(PhysicalColumn::new("ts_aware", 0)), + DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+07:00"))), + None, + )) as Arc; + + assert!(!is_time_index_expr(&projection, &aware_cast).unwrap()); + } + + #[test] + fn test_preserves_timestamp_values() { + let naive_nanosecond = DataType::Timestamp(TimeUnit::Nanosecond, None); + let naive_millisecond = DataType::Timestamp(TimeUnit::Millisecond, None); + let aware_nanosecond = DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+08:00"))); + let other_aware_nanosecond = + DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+07:00"))); + + // Keeping the unit and the values of a timestamp. + assert!(preserves_timestamp_values( + &naive_nanosecond, + &naive_nanosecond + )); + assert!(preserves_timestamp_values( + &aware_nanosecond, + &other_aware_nanosecond + )); + + // Changing the time unit, adding a timezone to a naive timestamp, or leaving the timestamp + // domain entirely all change the values of the timestamp. + assert!(!preserves_timestamp_values( + &naive_millisecond, + &naive_nanosecond + )); + assert!(!preserves_timestamp_values( + &aware_nanosecond, + &naive_nanosecond + )); + assert!(!preserves_timestamp_values( + &naive_nanosecond, + &DataType::Int64 + )); + assert!(!preserves_timestamp_values( + &DataType::Int64, + &naive_nanosecond + )); + + // Dropping an aware timestamp's timezone keeps the underlying values. Adding one to a + // naive timestamp is conservatively treated as non-preserving by the predicate. + assert!(preserves_timestamp_values( + &naive_nanosecond, + &aware_nanosecond + )); } #[test] @@ -524,7 +719,8 @@ mod test { DataType::Timestamp(TimeUnit::Millisecond, None), None, )) as Arc; - assert!(is_time_index_expr(&scan, &cast_expr).unwrap()); + // The cast changes the time unit (ns → ms), so it is no longer the time index. + assert!(!is_time_index_expr(&scan, &cast_expr).unwrap()); let non_timestamp_cast = Arc::new(CastExpr::new( Arc::new(PhysicalColumn::new("ts", 1)), @@ -626,13 +822,14 @@ mod test { } fn new_region_scan() -> Arc { + new_region_scan_with_unit(TimeUnit::Nanosecond) + } + + fn new_region_scan_with_unit(unit: TimeUnit) -> Arc { + let timestamp_type = ConcreteDataType::timestamp_datatype(unit.into()); let schema = Arc::new(Schema::new(vec![ ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false), - ColumnSchema::new( - "ts", - ConcreteDataType::timestamp_nanosecond_datatype(), - false, - ), + ColumnSchema::new("ts", timestamp_type.clone(), false), ])); let recordbatches = RecordBatches::try_new(schema.clone(), vec![]).unwrap(); let stream = recordbatches.as_stream(); @@ -649,11 +846,7 @@ mod test { column_id: 1, }) .push_column_metadata(ColumnMetadata { - column_schema: ColumnSchema::new( - "ts", - ConcreteDataType::timestamp_nanosecond_datatype(), - false, - ), + column_schema: ColumnSchema::new("ts", timestamp_type, false), semantic_type: SemanticType::Timestamp, column_id: 2, }); diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 481754e6df5..3b3dca0bbd0 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -949,6 +949,174 @@ ORDER BY t.key_id"#; ); } +/// Sorting by `CAST(ts AS TIMESTAMP(0))` must not be optimized as if it were sorting by the +/// (millisecond) time index column. +/// +/// The windowed sort rule still recognizes the cast as the time index column, but the working +/// ranges it derives from the region scan's `PartitionRange`s stay in the unit of the table's time +/// index (millisecond here). `PartSortExec` projects those ranges onto the unit of the sort column +/// while `WindowedSortExec` does not, so sorted runs (second) are checked against working ranges +/// (millisecond) and the range invariant trips with +/// "Current batch have data on the left side of working range, something is very wrong". +/// +/// The two rows are 1000 seconds apart, so casting the millisecond time index to second precision +/// truncates sub-second parts and collapses the first row onto second 0, which is before the start +/// of the region's time range. +/// +/// `to_timestamp_seconds(ts)` casts the time index to second precision *inside* the function, so it +/// has to be rejected exactly like the cast in the plan; the cast is just not visible as a +/// `CastExpr`. `to_timestamp_millis(ts)` keeps the values of the millisecond time index and is used +/// as the value-preserving control. +#[apply(standalone_instance_case)] +async fn test_windowed_sort_sort_column_cast_time_unit_mismatch(instance: Arc) { + let frontend = instance.frontend(); + + execute_sql( + &frontend, + r#"CREATE TABLE f05_mito_wc ( + ts TIMESTAMP(3) TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE + ) ENGINE = mito"#, + ) + .await; + + // The metric engine flavor only creates its physical table explicitly; rows are written to and + // read from the logical table built on it. + execute_sql( + &frontend, + r#"CREATE TABLE f05_metric_phy ( + ts TIMESTAMP(3) TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE + ) ENGINE = metric WITH ('physical_metric_table' = '')"#, + ) + .await; + execute_sql( + &frontend, + r#"CREATE TABLE f05_metric_wc ( + ts TIMESTAMP(3) TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE + ) ENGINE = metric WITH ('on_physical_table' = 'f05_metric_phy')"#, + ) + .await; + + let rows_by_time_index = r#"+-------------------------+------+ +| ts | host | ++-------------------------+------+ +| 1970-01-01T00:00:00.001 | a | +| 1970-01-01T00:16:40.001 | b | ++-------------------------+------+"#; + let rows_by_cast = r#"+---------------------+------+ +| t | host | ++---------------------+------+ +| 1970-01-01T00:00:00 | a | +| 1970-01-01T00:16:40 | b | ++---------------------+------+"#; + // `ORDER BY t DESC` returns the same rows in the opposite order; without distinct cast values + // the two rows tie and this expectation would be vacuous. `rows_by_cast` is an ascending + // fixture, so the descending query needs its own. + let rows_by_cast_desc = r#"+---------------------+------+ +| t | host | ++---------------------+------+ +| 1970-01-01T00:16:40 | b | +| 1970-01-01T00:00:00 | a | ++---------------------+------+"#; + + // `to_timestamp_seconds` casts the millisecond time index to second precision inside the + // function, so it returns the same rows as the direct cast above (the function is registered + // without a session time zone, so its result stays a naive timestamp). + let rows_by_seconds_wrapper = rows_by_cast; + let rows_by_seconds_wrapper_desc = rows_by_cast_desc; + // `to_timestamp_millis` keeps the values of the millisecond time index, so the control below + // returns the same rows as sorting by the time index itself, only under the `t` alias. + let rows_by_millis_wrapper = r#"+-------------------------+------+ +| t | host | ++-------------------------+------+ +| 1970-01-01T00:00:00.001 | a | +| 1970-01-01T00:16:40.001 | b | ++-------------------------+------+"#; + + for table in ["f05_mito_wc", "f05_metric_wc"] { + let output = execute_sql( + &frontend, + &format!("INSERT INTO {table} (ts, host, val) VALUES (1, 'a', 1), (1000001, 'b', 2)"), + ) + .await + .data; + assert!(matches!(output, OutputData::AffectedRows(2)), "{table}"); + + // Controls: sorting by the time index itself, applying a fetch, or sorting by the cast + // together with another column all keep working. + let output = execute_sql( + &frontend, + &format!("SELECT ts, host FROM {table} ORDER BY ts"), + ) + .await + .data; + check_output_stream(output, rows_by_time_index).await; + + let output = execute_sql( + &frontend, + &format!("SELECT to_timestamp_millis(ts) AS t, host FROM {table} ORDER BY t"), + ) + .await + .data; + check_output_stream(output, rows_by_millis_wrapper).await; + + let output = execute_sql( + &frontend, + &format!("SELECT CAST(ts AS TIMESTAMP(0)) AS t, host FROM {table} ORDER BY t LIMIT 10"), + ) + .await + .data; + check_output_stream(output, rows_by_cast).await; + + let output = execute_sql( + &frontend, + &format!("SELECT CAST(ts AS TIMESTAMP(0)) AS t, host FROM {table} ORDER BY t, host"), + ) + .await + .data; + check_output_stream(output, rows_by_cast).await; + } + + // Ordering by the cast or a function wrapper that casts inside the function returns the same + // rows as the controls above. + for table in ["f05_mito_wc", "f05_metric_wc"] { + for parallelism in [1, 4] { + for (sql, expected) in [ + ( + format!("SELECT CAST(ts AS TIMESTAMP(0)) AS t, host FROM {table} ORDER BY t"), + rows_by_cast, + ), + ( + format!( + "SELECT CAST(ts AS TIMESTAMP(0)) AS t, host FROM {table} ORDER BY t DESC" + ), + rows_by_cast_desc, + ), + ( + format!("SELECT to_timestamp_seconds(ts) AS t, host FROM {table} ORDER BY t"), + rows_by_seconds_wrapper, + ), + ( + format!( + "SELECT to_timestamp_seconds(ts) AS t, host FROM {table} ORDER BY t DESC" + ), + rows_by_seconds_wrapper_desc, + ), + ] { + let output = execute_sql_with_query_parallelism(&frontend, &sql, parallelism) + .await + .data; + check_output_stream(output, expected).await; + } + } + } +} + #[apply(both_instances_cases)] async fn test_show_create_table(instance: Arc) { let frontend = instance.frontend(); diff --git a/tests/cases/distributed/optimizer/windowed_sort_advance.result b/tests/cases/distributed/optimizer/windowed_sort_advance.result index 3c695f3879c..618a1547a14 100644 --- a/tests/cases/distributed/optimizer/windowed_sort_advance.result +++ b/tests/cases/distributed/optimizer/windowed_sort_advance.result @@ -107,7 +107,7 @@ EXPLAIN ANALYZE select to_timestamp_millis(ts) as ts, status, value from `a` whe |_|_|_| | 1_| 0_|_ProjectionExec: expr=[to_timestamp_millis(a.ts)@0 as ts, status@1 as status, value@2 as value] REDACTED |_|_|_SortPreservingMergeExec: [to_timestamp_millis(a.ts)@0 ASC NULLS LAST] REDACTED -|_|_|_WindowedSortExec: expr=to_timestamp_millis(a.ts)@0 ASC NULLS LAST num_ranges=REDACTED REDACTED +|_|_|_SortExec: expr=[to_timestamp_millis(a.ts)@0 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_ProjectionExec: expr=[to_timestamp_millis(ts@2) as to_timestamp_millis(a.ts), status@1 as status, value@0 as value] REDACTED |_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "projection": ["value", "status", "ts"], "filters": ["ts >= TimestampNanosecond(1773273600000000000, None)", "ts < TimestampNanosecond(1773273721000000000, None)"], "time_range":REDACTED |_|_|_| diff --git a/tests/cases/distributed/optimizer/windowed_sort_nanos.result b/tests/cases/distributed/optimizer/windowed_sort_nanos.result index 52d264073ae..122c6e64a29 100644 --- a/tests/cases/distributed/optimizer/windowed_sort_nanos.result +++ b/tests/cases/distributed/optimizer/windowed_sort_nanos.result @@ -67,7 +67,7 @@ EXPLAIN ANALYZE select to_timestamp_nanos(ts) as ts, status, value from `a_ms` w |_|_|_| | 1_| 0_|_ProjectionExec: expr=[to_timestamp_nanos(a_ms.ts)@0 as ts, status@1 as status, value@2 as value] REDACTED |_|_|_SortPreservingMergeExec: [to_timestamp_nanos(a_ms.ts)@0 ASC NULLS LAST] REDACTED -|_|_|_WindowedSortExec: expr=to_timestamp_nanos(a_ms.ts)@0 ASC NULLS LAST num_ranges=REDACTED REDACTED +|_|_|_SortExec: expr=[to_timestamp_nanos(a_ms.ts)@0 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_ProjectionExec: expr=[to_timestamp_nanos(ts@2) as to_timestamp_nanos(a_ms.ts), status@1 as status, value@0 as value] REDACTED |_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "projection": ["value", "status", "ts"], "filters": ["ts >= TimestampMillisecond(1773273600000, None)", "ts < TimestampMillisecond(1773273721000, None)"], "time_range":REDACTED |_|_|_| diff --git a/tests/cases/standalone/optimizer/windowed_sort_advance.result b/tests/cases/standalone/optimizer/windowed_sort_advance.result index f8863341471..123635de8c2 100644 --- a/tests/cases/standalone/optimizer/windowed_sort_advance.result +++ b/tests/cases/standalone/optimizer/windowed_sort_advance.result @@ -106,7 +106,7 @@ EXPLAIN ANALYZE select to_timestamp_millis(ts) as ts, status, value from `a` whe |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_SortPreservingMergeExec: [ts@0 ASC NULLS LAST] REDACTED -|_|_|_WindowedSortExec: expr=ts@0 ASC NULLS LAST num_ranges=REDACTED REDACTED +|_|_|_SortExec: expr=[ts@0 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_ProjectionExec: expr=[to_timestamp_millis(ts@2) as ts, status@1 as status, value@0 as value] REDACTED |_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "projection": ["value", "status", "ts"], "filters": ["ts >= TimestampNanosecond(1773273600000000000, None)", "ts < TimestampNanosecond(1773273721000000000, None)"], "time_range":REDACTED |_|_|_| diff --git a/tests/cases/standalone/optimizer/windowed_sort_nanos.result b/tests/cases/standalone/optimizer/windowed_sort_nanos.result index 9577b711704..3c808912b67 100644 --- a/tests/cases/standalone/optimizer/windowed_sort_nanos.result +++ b/tests/cases/standalone/optimizer/windowed_sort_nanos.result @@ -66,7 +66,7 @@ EXPLAIN ANALYZE select to_timestamp_nanos(ts) as ts, status, value from `a_ms` w |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_SortPreservingMergeExec: [ts@0 ASC NULLS LAST] REDACTED -|_|_|_WindowedSortExec: expr=ts@0 ASC NULLS LAST num_ranges=REDACTED REDACTED +|_|_|_SortExec: expr=[ts@0 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_ProjectionExec: expr=[to_timestamp_nanos(ts@2) as ts, status@1 as status, value@0 as value] REDACTED |_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "projection": ["value", "status", "ts"], "filters": ["ts >= TimestampMillisecond(1773273600000, None)", "ts < TimestampMillisecond(1773273721000, None)"], "time_range":REDACTED |_|_|_|