From 99ef93e14333cfd87bdc864dd4bf806d91c94e23 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Mon, 12 Oct 2026 05:10:18 +0000 Subject: [PATCH] fix(promql): follow IEEE 754 float semantics in comparisons and aggregations (#9479) * fix(promql): follow IEEE 754 float semantics in operators and aggregations - Comparisons: Arrow orders floats totally (NaN == NaN, NaN > +Inf); PromQL comparisons with NaN are false except !=. - min/max: skip NaN unless the whole group is NaN. DataFusion's grouped min/max let NaN replace and be replaced by neighbours, so the result depended on input order. - topk/bottomk rank NaN last. - round() rounds ties up and returns NaN for a zero step, like Prometheus. - sqrt() of a negative number is NaN instead of an error. Signed-off-by: Dennis Zhuang * fix(promql): round() without a step rounds to integers The planner passed a zero step for single-argument round(), which now yields NaN like Prometheus; pass 1 instead. Found by the compatibility harness. Signed-off-by: Dennis Zhuang * fix(promql): apply the sqrt guard to builtin functions and register min/max for plan decoding sqrt resolves through the session's builtin functions, not the UDF branch. Distributed plans are decoded with the serializer's session, which did not know prom_min/prom_max. Signed-off-by: Dennis Zhuang * test(promql): cover NaN and rounding semantics Signed-off-by: Dennis Zhuang * fix(promql): keep the nested count rewrite for PromQL min/max count(min(x) by (l)) was rewritten by count_nest_aggr only for the builtin min/max names. prom_min/prom_max return NULL for the same groups, so the rewrite stays valid. Update plan snapshots for the IEEE comparison, sqrt, topk and min/max changes. Signed-off-by: Dennis Zhuang * fix(promql): register prom_min/prom_max in the function registry Distributed plans split an aggregation into state and merge steps only for functions whose state function exists in FUNCTION_REGISTRY; prom_min and prom_max lived in the promql crate, so min/max stopped being pushed down to datanodes. Move them to common-function and register them before the state functions are derived. Signed-off-by: Dennis Zhuang * test: update sqlness results for PromQL float semantics Plans and column names now show prom_min/prom_max, the NaN guards of comparisons and sqrt, and the NaN sort key of topk. round() results follow Prometheus' formula. Distributed min/max keep the state/merge split. Signed-off-by: Dennis Zhuang * fix(promql): keep column names and NULLs in IEEE comparisons and sqrt Rewriting comparisons as `a op b AND NOT (isnan(a) OR isnan(b))` changed TQL output column names and turned NULL operands into true for `!=` and false for the bool forms. Implement comparisons and sqrt as PromQL UDFs that propagate NULL and report the column name of the expression they replace; prom_min/prom_max report min/max column names the same way. Signed-off-by: Dennis Zhuang * test: restore column names in sqlness results for PromQL float semantics Signed-off-by: Dennis Zhuang * test(promql): share the prom_min/prom_max groups accumulator with tests Signed-off-by: Dennis Zhuang * fix(promql): keep distinct names for prom_min/prom_max Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang (cherry picked from commit f3a0841f4f44c72534a55269e5dc3d7c94c5737c) Signed-off-by: WenyXu --- src/common/function/src/aggrs.rs | 1 + src/common/function/src/aggrs/extremum.rs | 259 +++++++++++++++++ src/common/function/src/function_registry.rs | 4 + src/promql/src/functions.rs | 2 + src/promql/src/functions/ieee_float.rs | 265 ++++++++++++++++++ src/promql/src/functions/round.rs | 62 ++-- src/query/src/optimizer/count_nest_aggr.rs | 19 +- src/query/src/promql/planner.rs | 115 ++++++-- src/query/src/promql/planner/test.rs | 55 +++- .../src/query_engine/default_serializer.rs | 6 +- .../explain/step_aggr_advance.result | 108 +++---- .../common/promql/count_values.result | 2 +- .../common/promql/delta_temporality.result | 2 +- .../common/promql/dotted_column_names.result | 14 +- .../common/promql/float_semantics.result | 115 ++++++++ .../common/promql/float_semantics.sql | 41 +++ .../standalone/common/promql/functions.result | 12 +- .../promql/tsid_binary_join_regression.result | 12 +- 18 files changed, 954 insertions(+), 140 deletions(-) create mode 100644 src/common/function/src/aggrs/extremum.rs create mode 100644 src/promql/src/functions/ieee_float.rs create mode 100644 tests/cases/standalone/common/promql/float_semantics.result create mode 100644 tests/cases/standalone/common/promql/float_semantics.sql diff --git a/src/common/function/src/aggrs.rs b/src/common/function/src/aggrs.rs index e618921a18d..bad38a075d4 100644 --- a/src/common/function/src/aggrs.rs +++ b/src/common/function/src/aggrs.rs @@ -14,6 +14,7 @@ pub mod approximate; pub mod count_hash; +pub mod extremum; #[cfg(feature = "geo")] pub mod geo; pub mod vector; diff --git a/src/common/function/src/aggrs/extremum.rs b/src/common/function/src/aggrs/extremum.rs new file mode 100644 index 00000000000..668c307ff36 --- /dev/null +++ b/src/common/function/src/aggrs/extremum.rs @@ -0,0 +1,259 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! PromQL `min` and `max` aggregation operators. +//! +//! PromQL ignores NaN samples unless every sample of a group is NaN. DataFusion's +//! `min`/`max` order NaN differently depending on the code path (its grouped +//! accumulator lets NaN replace the current value and be replaced by the next one), +//! so the result depends on the order of the input. + +use std::sync::Arc; + +use datafusion::arrow::array::{ArrayRef, AsArray}; +use datafusion::arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; +use datafusion::error::Result as DfResult; +use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; +use datafusion::logical_expr::{ + Accumulator, AggregateUDF, AggregateUDFImpl, GroupsAccumulator, Signature, Volatility, +}; +use datafusion_common::ScalarValue; +use datafusion_functions_aggregate_common::aggregate::groups_accumulator::prim_op::PrimitiveGroupsAccumulator; + +use crate::function_registry::FunctionRegistry; + +/// `max` or `min` with PromQL NaN semantics. +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct Extremum { + name: &'static str, + is_max: bool, + signature: Signature, +} + +impl Extremum { + pub const MAX_NAME: &'static str = "prom_max"; + pub const MIN_NAME: &'static str = "prom_min"; + + /// Registers both functions; the registry also derives their state functions, which + /// distributed plans need to split the aggregation between datanodes and the frontend. + pub fn register(registry: &FunctionRegistry) { + registry.register_aggr(Self::max_udaf()); + registry.register_aggr(Self::min_udaf()); + } + + pub fn max_udaf() -> AggregateUDF { + AggregateUDF::from(Self::new(Self::MAX_NAME, true)) + } + + pub fn min_udaf() -> AggregateUDF { + AggregateUDF::from(Self::new(Self::MIN_NAME, false)) + } + + fn new(name: &'static str, is_max: bool) -> Self { + Self { + name, + is_max, + signature: Signature::exact(vec![DataType::Float64], Volatility::Immutable), + } + } +} + +/// Folds `value` into `current` like Prometheus' aggregation: a NaN is replaced by any +/// sample, and a NaN sample never replaces a number. +fn fold(is_max: bool, current: &mut f64, value: f64) { + let better = if is_max { + value > *current + } else { + value < *current + }; + if current.is_nan() || better { + *current = value; + } +} + +impl AggregateUDFImpl for Extremum { + fn name(&self) -> &str { + self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> DfResult { + Ok(DataType::Float64) + } + + fn accumulator(&self, _acc_args: AccumulatorArgs) -> DfResult> { + Ok(Box::new(ExtremumAccumulator { + is_max: self.is_max, + value: None, + })) + } + + fn state_fields(&self, args: StateFieldsArgs) -> DfResult> { + Ok(vec![Arc::new(Field::new( + format!("{}[{}]", args.name, self.name), + DataType::Float64, + true, + ))]) + } + + fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool { + true + } + + fn create_groups_accumulator( + &self, + _args: AccumulatorArgs, + ) -> DfResult> { + Ok(groups_accumulator(self.is_max)) + } +} + +fn groups_accumulator(is_max: bool) -> Box { + Box::new( + PrimitiveGroupsAccumulator::::new(&DataType::Float64, move |cur, v| { + fold(is_max, cur, v) + }) + .with_starting_value(f64::NAN), + ) +} + +#[derive(Debug)] +struct ExtremumAccumulator { + is_max: bool, + value: Option, +} + +impl ExtremumAccumulator { + fn update(&mut self, values: &ArrayRef) { + for v in values.as_primitive::().iter().flatten() { + match &mut self.value { + Some(current) => fold(self.is_max, current, v), + None => self.value = Some(v), + } + } + } +} + +impl Accumulator for ExtremumAccumulator { + fn update_batch(&mut self, values: &[ArrayRef]) -> DfResult<()> { + self.update(&values[0]); + Ok(()) + } + + fn merge_batch(&mut self, states: &[ArrayRef]) -> DfResult<()> { + self.update(&states[0]); + Ok(()) + } + + fn state(&mut self) -> DfResult> { + Ok(vec![ScalarValue::Float64(self.value)]) + } + + fn evaluate(&mut self) -> DfResult { + Ok(ScalarValue::Float64(self.value)) + } + + fn size(&self) -> usize { + size_of_val(self) + } +} + +#[cfg(test)] +mod tests { + use datafusion::arrow::array::Float64Array; + use datafusion::logical_expr::EmitTo; + + use super::*; + + fn groups_result( + is_max: bool, + values: Vec>, + groups: Vec, + ) -> Vec> { + let num_groups = groups.iter().max().map_or(0, |g| g + 1); + let mut acc = groups_accumulator(is_max); + let values: ArrayRef = Arc::new(Float64Array::from(values)); + acc.update_batch(&[values], &groups, None, num_groups) + .unwrap(); + let out = acc.evaluate(EmitTo::All).unwrap(); + out.as_primitive::().iter().collect() + } + + fn is_nan(v: Option) -> bool { + v.is_some_and(f64::is_nan) + } + + #[test] + fn nan_is_ignored_unless_the_group_is_all_nan() { + // Group 0 has NaN on both sides of a number; group 1 is all NaN; group 2 has only NULL. + let values = vec![ + Some(f64::NAN), + Some(3.0), + Some(f64::NAN), + Some(f64::NAN), + None, + ]; + let groups = vec![0, 0, 0, 1, 2]; + for is_max in [true, false] { + let out = groups_result(is_max, values.clone(), groups.clone()); + assert_eq!(out[0], Some(3.0)); + assert!(is_nan(out[1])); + assert_eq!(out[2], None); + } + } + + #[test] + fn result_does_not_depend_on_input_order() { + let forward = vec![Some(1.0), Some(f64::NAN), Some(4.0), Some(2.0)]; + let mut backward = forward.clone(); + backward.reverse(); + assert_eq!( + groups_result(true, forward.clone(), vec![0; 4]), + vec![Some(4.0)] + ); + assert_eq!( + groups_result(true, backward.clone(), vec![0; 4]), + vec![Some(4.0)] + ); + assert_eq!(groups_result(false, forward, vec![0; 4]), vec![Some(1.0)]); + assert_eq!(groups_result(false, backward, vec![0; 4]), vec![Some(1.0)]); + } + + #[test] + fn merged_partial_states_keep_nan_semantics() { + let mut partial_nan = ExtremumAccumulator { + is_max: true, + value: None, + }; + partial_nan.update(&(Arc::new(Float64Array::from(vec![f64::NAN])) as ArrayRef)); + let mut partial_num = ExtremumAccumulator { + is_max: true, + value: None, + }; + partial_num.update(&(Arc::new(Float64Array::from(vec![2.0, 5.0])) as ArrayRef)); + + let mut merged = ExtremumAccumulator { + is_max: true, + value: None, + }; + for partial in [&mut partial_nan, &mut partial_num] { + let state = partial.state().unwrap()[0].to_array().unwrap(); + merged.merge_batch(&[state]).unwrap(); + } + assert_eq!(merged.evaluate().unwrap(), ScalarValue::Float64(Some(5.0))); + } +} diff --git a/src/common/function/src/function_registry.rs b/src/common/function/src/function_registry.rs index 6583f7ba607..8657636fe6b 100644 --- a/src/common/function/src/function_registry.rs +++ b/src/common/function/src/function_registry.rs @@ -25,6 +25,7 @@ use crate::admin::AdminFunction; use crate::aggrs::aggr_wrapper::StateMergeHelper; use crate::aggrs::approximate::ApproximateFunction; use crate::aggrs::count_hash::CountHash; +use crate::aggrs::extremum::Extremum; use crate::aggrs::vector::VectorFunction as VectorAggrFunction; use crate::function::{Function, FunctionRef}; use crate::function_factory::ScalarFunctionFactory; @@ -260,6 +261,9 @@ pub static FUNCTION_REGISTRY: LazyLock> = LazyLock::new(|| // CountHash function CountHash::register(&function_registry); + // PromQL min/max; registered before the state functions are derived. + Extremum::register(&function_registry); + // state function of supported aggregate functions StateMergeHelper::register(&function_registry); diff --git a/src/promql/src/functions.rs b/src/promql/src/functions.rs index 34618f0a953..6703505c277 100644 --- a/src/promql/src/functions.rs +++ b/src/promql/src/functions.rs @@ -19,6 +19,7 @@ mod double_exponential_smoothing; mod edge_count; mod extrapolate_rate; mod idelta; +mod ieee_float; mod native_histogram; mod predict_linear; mod quantile; @@ -45,6 +46,7 @@ pub use deriv::Deriv; pub use double_exponential_smoothing::DoubleExponentialSmoothing; pub use extrapolate_rate::{Delta, Increase, Rate}; pub use idelta::IDelta; +pub use ieee_float::{IeeeComparison, IeeeSqrt}; pub use native_histogram::{ MixedRange, NativeHistogramAbsentOverTime, NativeHistogramAdd, NativeHistogramAggAvg, NativeHistogramAggSum, NativeHistogramAvg, NativeHistogramAvgOverTime, NativeHistogramChanges, diff --git a/src/promql/src/functions/ieee_float.rs b/src/promql/src/functions/ieee_float.rs new file mode 100644 index 00000000000..d13480d8071 --- /dev/null +++ b/src/promql/src/functions/ieee_float.rs @@ -0,0 +1,265 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! PromQL comparison operators and `sqrt()`, which follow IEEE 754. +//! +//! Arrow compares floats by total order, where NaN equals NaN and is greater than `+Inf`, and +//! DataFusion's `sqrt` rejects negative input. In PromQL every comparison involving NaN is false +//! except `!=`, and the square root of a negative number is NaN. NULL input gives NULL output. +//! +//! The functions report the column name of the expression they replace (`a < b`, `sqrt(a)`), so +//! query output columns are unchanged. + +use std::sync::Arc; + +use datafusion::arrow::array::{AsArray, BooleanArray, Float64Array}; +use datafusion::arrow::datatypes::{DataType, Float64Type}; +use datafusion::error::Result as DfResult; +use datafusion::logical_expr::{ + ColumnarValue, Operator, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, +}; +use datafusion_expr::Expr; + +/// A PromQL comparison operator on two float operands. +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct IeeeComparison { + op: Operator, + signature: Signature, +} + +impl IeeeComparison { + /// The comparison UDF for `op`, which must be a comparison operator. + pub fn scalar_udf(op: Operator) -> ScalarUDF { + ScalarUDF::new_from_impl(Self { + op, + signature: Signature::uniform(2, vec![DataType::Float64], Volatility::Immutable), + }) + } + + /// All comparison UDFs, for registering them where plans are decoded. + pub fn all() -> Vec { + [ + Operator::Eq, + Operator::NotEq, + Operator::Lt, + Operator::LtEq, + Operator::Gt, + Operator::GtEq, + ] + .into_iter() + .map(Self::scalar_udf) + .collect() + } + + fn compare(&self, lhs: f64, rhs: f64) -> bool { + match self.op { + Operator::Eq => lhs == rhs, + Operator::NotEq => lhs != rhs, + Operator::Lt => lhs < rhs, + Operator::LtEq => lhs <= rhs, + Operator::Gt => lhs > rhs, + Operator::GtEq => lhs >= rhs, + _ => unreachable!("not a comparison operator: {}", self.op), + } + } +} + +impl ScalarUDFImpl for IeeeComparison { + fn name(&self) -> &str { + match self.op { + Operator::Eq => "prom_eq", + Operator::NotEq => "prom_ne", + Operator::Lt => "prom_lt", + Operator::LtEq => "prom_le", + Operator::Gt => "prom_gt", + Operator::GtEq => "prom_ge", + _ => unreachable!("not a comparison operator: {}", self.op), + } + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> DfResult { + Ok(DataType::Boolean) + } + + fn schema_name(&self, args: &[Expr]) -> DfResult { + Ok(format!( + "{} {} {}", + args[0].schema_name(), + self.op, + args[1].schema_name() + )) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DfResult { + let arrays = ColumnarValue::values_to_arrays(&args.args)?; + let lhs = arrays[0].as_primitive::(); + let rhs = arrays[1].as_primitive::(); + let result: BooleanArray = lhs + .iter() + .zip(rhs.iter()) + .map(|(lhs, rhs)| Some(self.compare(lhs?, rhs?))) + .collect(); + Ok(ColumnarValue::Array(Arc::new(result))) + } +} + +/// PromQL `sqrt()`. +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct IeeeSqrt { + signature: Signature, +} + +impl IeeeSqrt { + pub fn scalar_udf() -> ScalarUDF { + ScalarUDF::new_from_impl(Self { + signature: Signature::uniform(1, vec![DataType::Float64], Volatility::Immutable), + }) + } +} + +impl ScalarUDFImpl for IeeeSqrt { + fn name(&self) -> &str { + "prom_sqrt" + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> DfResult { + Ok(DataType::Float64) + } + + fn schema_name(&self, args: &[Expr]) -> DfResult { + Ok(format!("sqrt({})", args[0].schema_name())) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DfResult { + let arrays = ColumnarValue::values_to_arrays(&args.args)?; + let input = arrays[0].as_primitive::(); + let result: Float64Array = input.unary(f64::sqrt); + Ok(ColumnarValue::Array(Arc::new(result))) + } +} + +#[cfg(test)] +mod tests { + use datafusion::arrow::array::{Array, ArrayRef}; + use datafusion::arrow::datatypes::Field; + use datafusion_common::config::ConfigOptions; + + use super::*; + + fn invoke(udf: &ScalarUDF, args: Vec, return_type: DataType) -> ArrayRef { + let number_rows = args[0].len(); + let arg_fields = args + .iter() + .map(|array| Arc::new(Field::new("a", array.data_type().clone(), true))) + .collect(); + let result = udf + .invoke_with_args(ScalarFunctionArgs { + args: args.into_iter().map(ColumnarValue::Array).collect(), + arg_fields, + number_rows, + return_field: Arc::new(Field::new("r", return_type, true)), + config_options: Arc::new(ConfigOptions::default()), + }) + .unwrap(); + match result { + ColumnarValue::Array(array) => array, + ColumnarValue::Scalar(_) => unreachable!(), + } + } + + #[test] + fn comparisons_follow_ieee_and_propagate_null() { + let lhs: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(1.0), + Some(f64::NAN), + Some(f64::NAN), + None, + Some(f64::INFINITY), + ])); + let rhs: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(1.0), + Some(f64::NAN), + Some(1.0), + Some(f64::NAN), + Some(f64::NAN), + ])); + let expected = [ + ( + Operator::Eq, + [Some(true), Some(false), Some(false), None, Some(false)], + ), + ( + Operator::NotEq, + [Some(false), Some(true), Some(true), None, Some(true)], + ), + ( + Operator::Lt, + [Some(false), Some(false), Some(false), None, Some(false)], + ), + ( + Operator::LtEq, + [Some(true), Some(false), Some(false), None, Some(false)], + ), + ( + Operator::Gt, + [Some(false), Some(false), Some(false), None, Some(false)], + ), + ( + Operator::GtEq, + [Some(true), Some(false), Some(false), None, Some(false)], + ), + ]; + for (op, expected) in expected { + let result = invoke( + &IeeeComparison::scalar_udf(op), + vec![lhs.clone(), rhs.clone()], + DataType::Boolean, + ); + let result: Vec<_> = result.as_boolean().iter().collect(); + assert_eq!(result, expected, "{op}"); + } + } + + #[test] + fn sqrt_of_negative_is_nan() { + let input: ArrayRef = Arc::new(Float64Array::from(vec![Some(4.0), Some(-1.0), None])); + let result = invoke(&IeeeSqrt::scalar_udf(), vec![input], DataType::Float64); + let result = result.as_primitive::(); + assert_eq!(result.value(0), 2.0); + assert!(result.value(1).is_nan()); + assert!(result.is_null(2)); + } + + #[test] + fn column_names_match_the_replaced_expressions() { + let a = datafusion_expr::col("a"); + let b = datafusion_expr::lit(1.5_f64); + let comparison = + IeeeComparison::scalar_udf(Operator::LtEq).call(vec![a.clone(), b.clone()]); + assert_eq!( + comparison.schema_name().to_string(), + a.clone().lt_eq(b).schema_name().to_string() + ); + let sqrt = IeeeSqrt::scalar_udf().call(vec![a.clone()]); + assert_eq!(sqrt.schema_name().to_string(), "sqrt(a)"); + } +} diff --git a/src/promql/src/functions/round.rs b/src/promql/src/functions/round.rs index 72736ef6e2e..c5ba4c218c0 100644 --- a/src/promql/src/functions/round.rs +++ b/src/promql/src/functions/round.rs @@ -66,13 +66,7 @@ impl Round { let null_array = Float64Array::new_null(value_array.len()); return Ok(ColumnarValue::Array(Arc::new(null_array))); }; - let op = |a: f64| { - if nearest == 0.0 { - a.round() - } else { - (a / nearest).round() * nearest - } - }; + let op = |a: f64| round_to_nearest(a, nearest); let result: PrimitiveArray = value_array.as_primitive::().unary(op); Ok(ColumnarValue::Array(Arc::new(result) as _)) @@ -90,14 +84,10 @@ impl Round { )?; let result: PrimitiveArray = - datatypes::arrow::compute::binary(value_array, nearest_array, |a, nearest| { - if nearest == 0.0 { - a.round() - } else { - (a / nearest).round() * nearest - } - }) - .map_err(|err: ArrowError| DataFusionError::ArrowError(Box::new(err), None))?; + datatypes::arrow::compute::binary(value_array, nearest_array, round_to_nearest) + .map_err(|err: ArrowError| { + DataFusionError::ArrowError(Box::new(err), None) + })?; Ok(ColumnarValue::Array(Arc::new(result) as _)) } @@ -105,6 +95,13 @@ impl Round { } } +/// Rounds like Prometheus: ties round up (towards +Inf), and multiplying by the inverse +/// of `nearest` follows its float behavior, including NaN for `nearest == 0`. +fn round_to_nearest(value: f64, nearest: f64) -> f64 { + let inverse = 1.0 / nearest; + (value * inverse + 0.5).floor() / inverse +} + #[cfg(test)] mod tests { use datafusion_common::config::ConfigOptions; @@ -143,14 +140,43 @@ mod tests { #[test] fn test_round() { + // Expected values are Prometheus 3.15 `round()` results. test_round_f64(vec![123.456], 0.001, vec![123.456]); - test_round_f64(vec![123.456], 0.01, vec![123.46000000000001]); + test_round_f64(vec![123.456], 0.01, vec![123.46]); test_round_f64(vec![123.456], 0.1, vec![123.5]); - test_round_f64(vec![123.456], 0.0, vec![123.0]); test_round_f64(vec![123.456], 1.0, vec![123.0]); test_round_f64(vec![123.456], 10.0, vec![120.0]); test_round_f64(vec![123.456], 100.0, vec![100.0]); - test_round_f64(vec![123.456], 105.0, vec![105.0]); + test_round_f64(vec![123.456], 105.0, vec![104.99999999999999]); test_round_f64(vec![123.456], 1000.0, vec![0.0]); } + + #[test] + fn test_round_ties_towards_positive_infinity() { + test_round_f64(vec![-0.5], 1.0, vec![0.0]); + test_round_f64(vec![-2.5], 1.0, vec![-2.0]); + test_round_f64(vec![2.5], 1.0, vec![3.0]); + } + + #[test] + fn test_round_to_zero_is_nan() { + let round_udf = Round::scalar_udf(); + let input = vec![ + ColumnarValue::Array(Arc::new(Float64Array::from(vec![123.456]))), + ColumnarValue::Scalar(ScalarValue::Float64(Some(0.0))), + ]; + let arg_fields = vec![ + Arc::new(Field::new("a", input[0].data_type(), false)), + Arc::new(Field::new("b", input[1].data_type(), false)), + ]; + let args = ScalarFunctionArgs { + args: input, + arg_fields, + number_rows: 1, + return_field: Arc::new(Field::new("c", DataType::Float64, false)), + config_options: Arc::new(ConfigOptions::default()), + }; + let result = extract_array(&round_udf.invoke_with_args(args).unwrap()).unwrap(); + assert!(result.as_primitive::().value(0).is_nan()); + } } diff --git a/src/query/src/optimizer/count_nest_aggr.rs b/src/query/src/optimizer/count_nest_aggr.rs index 89ba4260742..6e3e9cdba05 100644 --- a/src/query/src/optimizer/count_nest_aggr.rs +++ b/src/query/src/optimizer/count_nest_aggr.rs @@ -15,6 +15,8 @@ use std::collections::HashSet; use std::sync::Arc; +use common_function::aggrs::aggr_wrapper::get_aggr_func; +use common_function::aggrs::extremum::Extremum; use datafusion::config::ConfigOptions; use datafusion::functions_aggregate::count::count_udaf; use datafusion::logical_expr::{Extension, LogicalPlan, LogicalPlanBuilder, Sort}; @@ -236,9 +238,8 @@ impl CountNestAggrRule { where F: FnOnce(&str) -> bool, { - let Expr::AggregateFunction(func) = expr else { - return None; - }; + // The PromQL planner aliases `prom_min`/`prom_max` to the builtin names. + let func = get_aggr_func(expr)?; let name = func.func.name(); if !accept_name(name) || func.params.filter.is_some() @@ -253,9 +254,19 @@ impl CountNestAggrRule { } fn is_supported_inner_aggregate(name: &str) -> bool { + // PromQL `min`/`max` plan to `prom_min`/`prom_max`, which return NULL for the same + // groups as the builtin ones. matches!( name, - "count" | "sum" | "avg" | "min" | "max" | "stddev_pop" | "var_pop" + "count" + | "sum" + | "avg" + | "min" + | "max" + | Extremum::MIN_NAME + | Extremum::MAX_NAME + | "stddev_pop" + | "var_pop" ) } diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 972f04b9ccf..afe52c7b560 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -27,6 +27,7 @@ use async_recursion::async_recursion; use catalog::table_source::DfTableSourceProvider; use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; +use common_function::aggrs::extremum::Extremum; use common_function::function::FunctionContext; use common_query::native_histogram::native_histogram_value_type; use common_query::prelude::{ @@ -69,18 +70,18 @@ use promql::extension_plan::{ }; use promql::functions::{ AbsentOverTime, AvgOverTime, Changes, CountOverTime, Delta, Deriv, DoubleExponentialSmoothing, - IDelta, Increase, LastOverTime, MatchGroupViolation, MaxOverTime, MinOverTime, MixedRange, - NativeHistogramAbsentOverTime, NativeHistogramAdd, NativeHistogramAggAvg, - NativeHistogramAggSum, NativeHistogramAvg, NativeHistogramAvgOverTime, NativeHistogramChanges, - NativeHistogramCount, NativeHistogramCountOverTime, NativeHistogramDelta, - NativeHistogramDivScalar, NativeHistogramDrop, NativeHistogramEq, NativeHistogramIDelta, - NativeHistogramIRate, NativeHistogramIncrease, NativeHistogramLastOverTime, - NativeHistogramMulScalar, NativeHistogramNeg, NativeHistogramNotEq, - NativeHistogramPresentOverTime, NativeHistogramRate, NativeHistogramResets, - NativeHistogramScalarMul, NativeHistogramStddev, NativeHistogramStdvar, NativeHistogramSub, - NativeHistogramSum, NativeHistogramSumOverTime, NativeHistogramToString, PredictLinear, - PresentOverTime, PromqlFloatToString, QuantileOverTime, Rate, Resets, Round, StddevOverTime, - StdvarOverTime, SumOverTime, UniqueMatchGroup, quantile_udaf, + IDelta, IeeeComparison, IeeeSqrt, Increase, LastOverTime, MatchGroupViolation, MaxOverTime, + MinOverTime, MixedRange, NativeHistogramAbsentOverTime, NativeHistogramAdd, + NativeHistogramAggAvg, NativeHistogramAggSum, NativeHistogramAvg, NativeHistogramAvgOverTime, + NativeHistogramChanges, NativeHistogramCount, NativeHistogramCountOverTime, + NativeHistogramDelta, NativeHistogramDivScalar, NativeHistogramDrop, NativeHistogramEq, + NativeHistogramIDelta, NativeHistogramIRate, NativeHistogramIncrease, + NativeHistogramLastOverTime, NativeHistogramMulScalar, NativeHistogramNeg, + NativeHistogramNotEq, NativeHistogramPresentOverTime, NativeHistogramRate, + NativeHistogramResets, NativeHistogramScalarMul, NativeHistogramStddev, NativeHistogramStdvar, + NativeHistogramSub, NativeHistogramSum, NativeHistogramSumOverTime, NativeHistogramToString, + PredictLinear, PresentOverTime, PromqlFloatToString, QuantileOverTime, Rate, Resets, Round, + StddevOverTime, StdvarOverTime, SumOverTime, UniqueMatchGroup, quantile_udaf, }; use promql_parser::label::{METRIC_NAME, MatchOp, Matcher, Matchers}; use promql_parser::parser::token::TokenType; @@ -3963,10 +3964,14 @@ impl PromPlanner { } "round" => { if other_input_exprs.is_empty() { - other_input_exprs.push_front(0.0f64.lit()); + other_input_exprs.push_front(1.0f64.lit()); } ScalarFunc::DataFusionUdf(Arc::new(Round::scalar_udf())) } + // DataFusion's `sqrt` rejects negative input; PromQL returns NaN for it. + "sqrt" if !self.all_field_columns_are_native_histograms(input_schema) => { + ScalarFunc::DataFusionUdf(Arc::new(IeeeSqrt::scalar_udf())) + } "rad" | "deg" | "sgn" if self.all_field_columns_are_native_histograms(input_schema) => { ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name)) } @@ -4674,19 +4679,41 @@ impl PromPlanner { vec![] }; - // update value column name according to the aggregators, - let mut new_field_columns = Vec::with_capacity(self.ctx.field_columns.len()); + // The alias is built from normalized columns so it matches the builtin's + // qualified name. + let exprs = normalize_cols(exprs, input_plan) + .context(DataFusionPlanningSnafu)? + .into_iter() + .map(Self::alias_extremum_as_builtin) + .collect::>(); - let normalized_exprs = - normalize_cols(exprs.iter().cloned(), input_plan).context(DataFusionPlanningSnafu)?; - for expr in normalized_exprs { - new_field_columns.push(expr.schema_name().to_string()); - } - self.ctx.field_columns = new_field_columns; + // update value column name according to the aggregators, + self.ctx.field_columns = exprs + .iter() + .map(|expr| expr.schema_name().to_string()) + .collect(); Ok((exprs, prev_field_exprs)) } + /// `prom_min`/`prom_max` keep their own names so SQL can use them next to `min`/`max` + /// in one aggregate. PromQL output columns keep the builtin names. + fn alias_extremum_as_builtin(expr: DfExpr) -> DfExpr { + let DfExpr::AggregateFunction(aggr) = &expr else { + return expr; + }; + let builtin = match aggr.func.name() { + Extremum::MIN_NAME => min_udaf(), + Extremum::MAX_NAME => max_udaf(), + _ => return expr, + }; + let name = builtin + .call(aggr.params.args.clone()) + .schema_name() + .to_string(); + expr.alias(name) + } + fn create_numeric_aggregate_expr( op: TokenType, param: &Option>, @@ -4704,8 +4731,8 @@ impl PromPlanner { } token::T_AVG => avg_udaf().call(vec![input]), token::T_COUNT_VALUES | token::T_COUNT => count_udaf().call(vec![input]), - token::T_MIN => min_udaf().call(vec![input]), - token::T_MAX => max_udaf().call(vec![input]), + token::T_MIN => Extremum::min_udaf().call(vec![input]), + token::T_MAX => Extremum::max_udaf().call(vec![input]), // PromQL's `group()` aggregator produces 1 for each group. // Use `max(1.0)` (per-group) to match semantics and output type (Float64). token::T_GROUP => max_udaf().call(vec![lit(1_f64)]), @@ -5049,7 +5076,15 @@ impl PromPlanner { .field_columns .iter() .map(|col| { - let mut sort_exprs = Vec::with_capacity(self.ctx.tag_columns.len() + 1); + let mut sort_exprs = Vec::with_capacity(self.ctx.tag_columns.len() + 2); + // PromQL ranks NaN last for both `topk` and `bottomk`, while DataFusion + // orders NaN above every number. + sort_exprs.push( + datafusion_functions::math::expr_fn::isnan(DfExpr::Column(Column::from_name( + col, + ))) + .sort(true, true), + ); // Order by value in the specific order sort_exprs.push(DfExpr::Column(Column::from_name(col)).sort(asc, true)); // Then tags if the values are equal, @@ -5240,12 +5275,24 @@ impl PromPlanner { token::T_MOD => Ok(Box::new(move |lhs: DfExpr, rhs| { Ok(cast_float(lhs) % cast_float(rhs)) })), - token::T_EQLC => Ok(Box::new(|lhs, rhs| Ok(lhs.eq(rhs)))), - token::T_NEQ => Ok(Box::new(|lhs, rhs| Ok(lhs.not_eq(rhs)))), - token::T_GTR => Ok(Box::new(|lhs, rhs| Ok(lhs.gt(rhs)))), - token::T_LSS => Ok(Box::new(|lhs, rhs| Ok(lhs.lt(rhs)))), - token::T_GTE => Ok(Box::new(|lhs, rhs| Ok(lhs.gt_eq(rhs)))), - token::T_LTE => Ok(Box::new(|lhs, rhs| Ok(lhs.lt_eq(rhs)))), + token::T_EQLC => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::Eq, rhs)) + })), + token::T_NEQ => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::NotEq, rhs)) + })), + token::T_GTR => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::Gt, rhs)) + })), + token::T_LSS => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::Lt, rhs)) + })), + token::T_GTE => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::GtEq, rhs)) + })), + token::T_LTE => Ok(Box::new(|lhs, rhs| { + Ok(Self::ieee_comparison(lhs, Operator::LtEq, rhs)) + })), token::T_POW => Ok(Box::new(move |lhs, rhs| { Ok(DfExpr::ScalarFunction(ScalarFunction { func: datafusion_functions::math::power(), @@ -5262,6 +5309,14 @@ impl PromPlanner { } } + /// Compares two expressions with IEEE 754 semantics, as PromQL does; see [`IeeeComparison`]. + fn ieee_comparison(lhs: DfExpr, op: Operator, rhs: DfExpr) -> DfExpr { + DfExpr::ScalarFunction(ScalarFunction { + func: Arc::new(IeeeComparison::scalar_udf(op)), + args: vec![lhs, rhs], + }) + } + /// Check if the given op is a [comparison operator](https://prometheus.io/docs/prometheus/latest/querying/operators/#comparison-binary-operators). fn is_token_a_comparison_op(token: TokenType) -> bool { matches!( diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index f78162d8fdb..ae40640b50b 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -1679,7 +1679,18 @@ async fn single_sort_desc() { #[tokio::test] async fn single_sqrt() { - do_single_instant_function_call("sqrt", "sqrt").await; + // Negative input is mapped to NaN before DataFusion's `sqrt`, which rejects it. + indie_query_plan_compare( + r#"sqrt(some_metric{tag_0!="bar"})"#, + "Filter: sqrt(field_0) IS NOT NULL [timestamp:Timestamp(ms), sqrt(field_0):Float64;N, tag_0:Utf8]\ + \n Projection: some_metric.timestamp, prom_sqrt(some_metric.field_0) AS sqrt(field_0), some_metric.tag_0 [timestamp:Timestamp(ms), sqrt(field_0):Float64;N, tag_0:Utf8]\ + \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ + \n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ + \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ + \n Filter: some_metric.tag_0 != Utf8(\"bar\") AND some_metric.timestamp >= TimestampMillisecond(-999, None) AND some_metric.timestamp <= TimestampMillisecond(100000000, None) [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ + \n TableScan: some_metric [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]", + ) + .await; } #[tokio::test] @@ -1811,6 +1822,19 @@ async fn single_rad() { // }, // }, async fn do_aggregate_expr_plan(fn_name: &str, plan_name: &str) { + do_aggregate_expr_plan_with_display(fn_name, plan_name, plan_name).await; +} + +/// `display_name` is the aggregate function as the plan shows it, `plan_name` the name of the +/// output column; they differ for PromQL `min`/`max`. +async fn do_aggregate_expr_plan_with_display(fn_name: &str, plan_name: &str, display_name: &str) { + let display = |field: &str| { + if display_name == plan_name { + format!("{plan_name}({field})") + } else { + format!("{display_name}({field}) AS {plan_name}({field})") + } + }; let prom_expr = parser::parse(&format!( "{fn_name} by (tag_1)(some_metric{{tag_0!=\"bar\"}})", )) @@ -1837,13 +1861,16 @@ async fn do_aggregate_expr_plan(fn_name: &str, plan_name: &str) { .unwrap(); let expected_no_without = String::from( "Sort: some_metric.tag_1 ASC NULLS LAST, some_metric.timestamp ASC NULLS LAST [tag_1:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ - \n Aggregate: groupBy=[[some_metric.tag_1, some_metric.timestamp]], aggr=[[TEMPLATE(some_metric.field_0), TEMPLATE(some_metric.field_1)]] [tag_1:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ + \n Aggregate: groupBy=[[some_metric.tag_1, some_metric.timestamp]], aggr=[[DISPLAY_0, DISPLAY_1]] [tag_1:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n PromSeriesDivide: tags=[\"tag_0\", \"tag_1\"] [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.tag_1 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n Filter: some_metric.tag_0 != Utf8(\"bar\") AND some_metric.timestamp >= TimestampMillisecond(-999, None) AND some_metric.timestamp <= TimestampMillisecond(100000000, None) [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n TableScan: some_metric [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]" - ).replace("TEMPLATE", plan_name); + ) + .replace("TEMPLATE", plan_name) + .replace("DISPLAY_0", &display("some_metric.field_0")) + .replace("DISPLAY_1", &display("some_metric.field_1")); assert_eq!( plan.display_indent_schema().to_string(), expected_no_without @@ -1866,13 +1893,15 @@ async fn do_aggregate_expr_plan(fn_name: &str, plan_name: &str) { .unwrap(); let expected_without = String::from( "Sort: some_metric.tag_0 ASC NULLS LAST, some_metric.timestamp ASC NULLS LAST [tag_0:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ - \n Aggregate: groupBy=[[some_metric.tag_0, some_metric.timestamp]], aggr=[[TEMPLATE(some_metric.field_0), TEMPLATE(some_metric.field_1)]] [tag_0:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ + \n Aggregate: groupBy=[[some_metric.tag_0, some_metric.timestamp]], aggr=[[DISPLAY_0, DISPLAY_1]] [tag_0:Utf8, timestamp:Timestamp(ms), TEMPLATE(some_metric.field_0):Float64;N, TEMPLATE(some_metric.field_1):Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n PromSeriesDivide: tags=[\"tag_0\", \"tag_1\"] [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.tag_1 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n Filter: some_metric.tag_0 != Utf8(\"bar\") AND some_metric.timestamp >= TimestampMillisecond(-999, None) AND some_metric.timestamp <= TimestampMillisecond(100000000, None) [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]\ \n TableScan: some_metric [tag_0:Utf8, tag_1:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, field_1:Float64;N]" - ).replace("TEMPLATE", plan_name); + ).replace("TEMPLATE", plan_name) + .replace("DISPLAY_0", &display("some_metric.field_0")) + .replace("DISPLAY_1", &display("some_metric.field_1")); assert_eq!(plan.display_indent_schema().to_string(), expected_without); } @@ -3898,12 +3927,12 @@ async fn aggregate_count() { #[tokio::test] async fn aggregate_min() { - do_aggregate_expr_plan("min", "min").await; + do_aggregate_expr_plan_with_display("min", "min", "prom_min").await; } #[tokio::test] async fn aggregate_max() { - do_aggregate_expr_plan("max", "max").await; + do_aggregate_expr_plan_with_display("max", "max", "prom_max").await; } #[tokio::test] @@ -4081,7 +4110,7 @@ async fn binary_op_literal_literal() { async fn simple_bool_grammar() { let query = "some_metric != bool 1.2345"; let expected = String::from( - "Projection: some_metric.tag_0, some_metric.timestamp, CAST(some_metric.field_0 != Float64(1.2345) AS Float64) AS field_0 != Float64(1.2345) [tag_0:Utf8, timestamp:Timestamp(ms), field_0 != Float64(1.2345):Float64;N]\ + "Projection: some_metric.tag_0, some_metric.timestamp, CAST(prom_ne(some_metric.field_0, Float64(1.2345)) AS Float64) AS field_0 != Float64(1.2345) [tag_0:Utf8, timestamp:Timestamp(ms), field_0 != Float64(1.2345):Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ @@ -4096,7 +4125,7 @@ async fn simple_bool_grammar() { async fn bool_with_additional_arithmetic() { let query = "some_metric + (1 == bool 2)"; let expected = String::from( - "Projection: some_metric.tag_0, some_metric.timestamp, CAST(some_metric.field_0 AS Float64) + CAST(Float64(1) = Float64(2) AS Float64) AS field_0 + Float64(1) = Float64(2) [tag_0:Utf8, timestamp:Timestamp(ms), field_0 + Float64(1) = Float64(2):Float64;N]\ + "Projection: some_metric.tag_0, some_metric.timestamp, CAST(some_metric.field_0 AS Float64) + CAST(prom_eq(Float64(1), Float64(2)) AS Float64) AS field_0 + Float64(1) = Float64(2) [tag_0:Utf8, timestamp:Timestamp(ms), field_0 + Float64(1) = Float64(2):Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ @@ -5064,7 +5093,7 @@ async fn native_histogram_mixed_field_table_behaves() { async fn less_filter_on_value() { let query = "some_metric < 1.2345"; let expected = String::from( - "Filter: some_metric.field_0 < Float64(1.2345) [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ + "Filter: prom_lt(some_metric.field_0, Float64(1.2345)) [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ \n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\ @@ -6642,9 +6671,9 @@ async fn test_topk_expr() { .await .unwrap(); let expected = "Projection: sum(prometheus_tsdb_head_series.greptime_value), prometheus_tsdb_head_series.ip, prometheus_tsdb_head_series.greptime_timestamp [sum(prometheus_tsdb_head_series.greptime_value):Float64;N, ip:Utf8, greptime_timestamp:Timestamp(ms)]\ - \n Sort: prometheus_tsdb_head_series.greptime_timestamp ASC NULLS LAST, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ASC NULLS LAST [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ - \n Filter: row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW <= Float64(10) [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ - \n WindowAggr: windowExpr=[[row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ + \n Sort: prometheus_tsdb_head_series.greptime_timestamp ASC NULLS LAST, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ASC NULLS LAST [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ + \n Filter: row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW <= Float64(10) [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ + \n WindowAggr: windowExpr=[[row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]] [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N, row_number() PARTITION BY [prometheus_tsdb_head_series.greptime_timestamp] ORDER BY [isnan(sum(prometheus_tsdb_head_series.greptime_value)) ASC NULLS FIRST, sum(prometheus_tsdb_head_series.greptime_value) DESC NULLS FIRST, prometheus_tsdb_head_series.ip DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:UInt64]\ \n Sort: prometheus_tsdb_head_series.ip ASC NULLS LAST, prometheus_tsdb_head_series.greptime_timestamp ASC NULLS LAST [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N]\ \n Aggregate: groupBy=[[prometheus_tsdb_head_series.ip, prometheus_tsdb_head_series.greptime_timestamp]], aggr=[[sum(prometheus_tsdb_head_series.greptime_value)]] [ip:Utf8, greptime_timestamp:Timestamp(ms), sum(prometheus_tsdb_head_series.greptime_value):Float64;N]\ \n PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[greptime_timestamp] [ip:Utf8, greptime_timestamp:Timestamp(ms), greptime_value:Float64;N]\ diff --git a/src/query/src/query_engine/default_serializer.rs b/src/query/src/query_engine/default_serializer.rs index f4d6b29fa2c..53887d78f75 100644 --- a/src/query/src/query_engine/default_serializer.rs +++ b/src/query/src/query_engine/default_serializer.rs @@ -30,7 +30,7 @@ use datafusion_expr::UserDefinedLogicalNode; use greptime_proto::substrait_extension::MergeScan as PbMergeScan; use promql::functions::{ AbsentOverTime, AvgOverTime, Changes, CountOverTime, Delta, Deriv, DoubleExponentialSmoothing, - IDelta, Increase, LastOverTime, MaxOverTime, MinOverTime, MixedRange, + IDelta, IeeeComparison, IeeeSqrt, Increase, LastOverTime, MaxOverTime, MinOverTime, MixedRange, NativeHistogramAbsentOverTime, NativeHistogramAdd, NativeHistogramAggAvg, NativeHistogramAggSum, NativeHistogramAvg, NativeHistogramAvgOverTime, NativeHistogramChanges, NativeHistogramCount, NativeHistogramCountOverTime, NativeHistogramDelta, @@ -151,6 +151,10 @@ impl SubstraitPlanDecoder for DefaultPlanDecoder { } let _ = session_state.register_udaf(quantile_udaf()); + for udf in IeeeComparison::all() { + let _ = session_state.register_udf(Arc::new(udf)); + } + let _ = session_state.register_udf(Arc::new(IeeeSqrt::scalar_udf())); let _ = session_state.register_udf(Arc::new(IDelta::::scalar_udf())); let _ = session_state.register_udf(Arc::new(IDelta::::scalar_udf())); diff --git a/tests/cases/distributed/explain/step_aggr_advance.result b/tests/cases/distributed/explain/step_aggr_advance.result index 348021360ff..11adf22f866 100644 --- a/tests/cases/distributed/explain/step_aggr_advance.result +++ b/tests/cases/distributed/explain/step_aggr_advance.result @@ -21,26 +21,26 @@ Affected Rows: 0 -- SQLNESS REPLACE (Hash.*) REDACTED tql explain (1752591864, 1752592164, '30s') max by (a, b, c) (max_over_time(aggr_optimize_not [2m])); -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | MergeSort: aggr_optimize_not.a ASC NULLS LAST, aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Sort: aggr_optimize_not.a ASC NULLS LAST, aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | -| | Aggregate: groupBy=[[aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.greptime_timestamp]], aggr=[[max(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | -| | Filter: prom_max_over_time(greptime_timestamp_range,greptime_value) IS NOT NULL | -| | Projection: aggr_optimize_not.greptime_timestamp, prom_max_over_time(greptime_timestamp_range, greptime_value) AS prom_max_over_time(greptime_timestamp_range,greptime_value), aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d | -| | PromRangeManipulate: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp], values=["greptime_value"] | -| | PromSeriesNormalize: offset=[0], time index=[greptime_timestamp], filter NaN: [true] | -| | PromSeriesDivide: tags=["a", "b", "c", "d"] | -| | Sort: aggr_optimize_not.a ASC NULLS FIRST, aggr_optimize_not.b ASC NULLS FIRST, aggr_optimize_not.c ASC NULLS FIRST, aggr_optimize_not.d ASC NULLS FIRST, aggr_optimize_not.greptime_timestamp ASC NULLS FIRST | -| | Filter: aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None) AND aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None) | -| | TableScan: aggr_optimize_not, partial_filters=[aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None), aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None)] | -| | ]] | -| physical_plan | MergeSortExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | MergeSort: aggr_optimize_not.a ASC NULLS LAST, aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Sort: aggr_optimize_not.a ASC NULLS LAST, aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | +| | Aggregate: groupBy=[[aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.greptime_timestamp]], aggr=[[prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value)) AS max(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | +| | Filter: prom_max_over_time(greptime_timestamp_range,greptime_value) IS NOT NULL | +| | Projection: aggr_optimize_not.greptime_timestamp, prom_max_over_time(greptime_timestamp_range, greptime_value) AS prom_max_over_time(greptime_timestamp_range,greptime_value), aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d | +| | PromRangeManipulate: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp], values=["greptime_value"] | +| | PromSeriesNormalize: offset=[0], time index=[greptime_timestamp], filter NaN: [true] | +| | PromSeriesDivide: tags=["a", "b", "c", "d"] | +| | Sort: aggr_optimize_not.a ASC NULLS FIRST, aggr_optimize_not.b ASC NULLS FIRST, aggr_optimize_not.c ASC NULLS FIRST, aggr_optimize_not.d ASC NULLS FIRST, aggr_optimize_not.greptime_timestamp ASC NULLS FIRST | +| | Filter: aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None) AND aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None) | +| | TableScan: aggr_optimize_not, partial_filters=[aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None), aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None)] | +| | ]] | +| physical_plan | MergeSortExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] | | | MergeScanExec: REDACTED -| | | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| | | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS REPLACE (metrics.*) REDACTED -- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED @@ -57,11 +57,12 @@ tql analyze (1752591864, 1752592164, '30s') max by (a, b, c) (max_over_time(aggr | 0_| 0_|_MergeSortExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| -| 1_| 0_|_SortPreservingMergeExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED +| 1_| 0_|_ProjectionExec: expr=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp, prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))@4 as max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_SortPreservingMergeExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED |_|_|_SortExec: expr=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST], preserve_partitioning=[true] REDACTED -|_|_|_AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp], aggr=[max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp], aggr=[prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_AggregateExec: mode=Partial, gby=[a@2 as a, b@3 as b, c@4 as c, greptime_timestamp@0 as greptime_timestamp], aggr=[max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[a@2 as a, b@3 as b, c@4 as c, greptime_timestamp@0 as greptime_timestamp], aggr=[prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_FilterExec: prom_max_over_time(greptime_timestamp_range,greptime_value)@1 IS NOT NULL REDACTED |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, prom_max_over_time(greptime_timestamp_range@6, greptime_value@5) as prom_max_over_time(greptime_timestamp_range,greptime_value), a@0 as a, b@1 as b, c@2 as c] REDACTED |_|_|_PromRangeManipulateExec: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp] REDACTED @@ -69,11 +70,12 @@ tql analyze (1752591864, 1752592164, '30s') max by (a, b, c) (max_over_time(aggr |_|_|_PromSeriesDivideExec: tags=["a", "b", "c", "d"] REDACTED |_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED |_|_|_| -| 1_| 1_|_SortPreservingMergeExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED +| 1_| 1_|_ProjectionExec: expr=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp, prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))@4 as max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_SortPreservingMergeExec: [a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED |_|_|_SortExec: expr=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST], preserve_partitioning=[true] REDACTED -|_|_|_AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp], aggr=[max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c, greptime_timestamp@3 as greptime_timestamp], aggr=[prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_AggregateExec: mode=Partial, gby=[a@2 as a, b@3 as b, c@4 as c, greptime_timestamp@0 as greptime_timestamp], aggr=[max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[a@2 as a, b@3 as b, c@4 as c, greptime_timestamp@0 as greptime_timestamp], aggr=[prom_max(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_FilterExec: prom_max_over_time(greptime_timestamp_range,greptime_value)@1 IS NOT NULL REDACTED |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, prom_max_over_time(greptime_timestamp_range@6, greptime_value@5) as prom_max_over_time(greptime_timestamp_range,greptime_value), a@0 as a, b@1 as b, c@2 as c] REDACTED |_|_|_PromRangeManipulateExec: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp] REDACTED @@ -305,31 +307,31 @@ tql analyze (1752591864, 1752592164, '30s') count by (a, b, c, d) (max_over_time -- SQLNESS REPLACE (Hash.*) REDACTED tql explain (1752591864, 1752592164, '30s') min by (b, c, d) (max_over_time(aggr_optimize_not [2m])); -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.d ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | -| | Aggregate: groupBy=[[aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d, aggr_optimize_not.greptime_timestamp]], aggr=[[__min_merge(__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) AS min(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Aggregate: groupBy=[[aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d, aggr_optimize_not.greptime_timestamp]], aggr=[[__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | -| | Filter: prom_max_over_time(greptime_timestamp_range,greptime_value) IS NOT NULL | -| | Projection: aggr_optimize_not.greptime_timestamp, prom_max_over_time(greptime_timestamp_range, greptime_value) AS prom_max_over_time(greptime_timestamp_range,greptime_value), aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d | -| | PromRangeManipulate: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp], values=["greptime_value"] | -| | PromSeriesNormalize: offset=[0], time index=[greptime_timestamp], filter NaN: [true] | -| | PromSeriesDivide: tags=["a", "b", "c", "d"] | -| | Sort: aggr_optimize_not.a ASC NULLS FIRST, aggr_optimize_not.b ASC NULLS FIRST, aggr_optimize_not.c ASC NULLS FIRST, aggr_optimize_not.d ASC NULLS FIRST, aggr_optimize_not.greptime_timestamp ASC NULLS FIRST | -| | Filter: aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None) AND aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None) | -| | TableScan: aggr_optimize_not, partial_filters=[aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None), aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None)] | -| | ]] | -| physical_plan | SortPreservingMergeExec: [b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] | -| | SortExec: expr=[b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST], preserve_partitioning=[true] | -| | AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_merge(__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] | ++---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: aggr_optimize_not.b ASC NULLS LAST, aggr_optimize_not.c ASC NULLS LAST, aggr_optimize_not.d ASC NULLS LAST, aggr_optimize_not.greptime_timestamp ASC NULLS LAST | +| | Aggregate: groupBy=[[aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d, aggr_optimize_not.greptime_timestamp]], aggr=[[__prom_min_merge(__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) AS min(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Aggregate: groupBy=[[aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d, aggr_optimize_not.greptime_timestamp]], aggr=[[__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))]] | +| | Filter: prom_max_over_time(greptime_timestamp_range,greptime_value) IS NOT NULL | +| | Projection: aggr_optimize_not.greptime_timestamp, prom_max_over_time(greptime_timestamp_range, greptime_value) AS prom_max_over_time(greptime_timestamp_range,greptime_value), aggr_optimize_not.a, aggr_optimize_not.b, aggr_optimize_not.c, aggr_optimize_not.d | +| | PromRangeManipulate: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp], values=["greptime_value"] | +| | PromSeriesNormalize: offset=[0], time index=[greptime_timestamp], filter NaN: [true] | +| | PromSeriesDivide: tags=["a", "b", "c", "d"] | +| | Sort: aggr_optimize_not.a ASC NULLS FIRST, aggr_optimize_not.b ASC NULLS FIRST, aggr_optimize_not.c ASC NULLS FIRST, aggr_optimize_not.d ASC NULLS FIRST, aggr_optimize_not.greptime_timestamp ASC NULLS FIRST | +| | Filter: aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None) AND aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None) | +| | TableScan: aggr_optimize_not, partial_filters=[aggr_optimize_not.greptime_timestamp >= TimestampMillisecond(1752591744001, None), aggr_optimize_not.greptime_timestamp <= TimestampMillisecond(1752592164000, None)] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] | +| | SortExec: expr=[b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_merge(__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] | | | RepartitionExec: partitioning=REDACTED -| | AggregateExec: mode=Partial, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_merge(__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] | +| | AggregateExec: mode=Partial, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_merge(__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] | | | RepartitionExec: partitioning=REDACTED | | MergeScanExec: REDACTED -| | | -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| | | ++---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS REPLACE (metrics.*) REDACTED -- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED @@ -345,15 +347,15 @@ tql analyze (1752591864, 1752592164, '30s') min by (b, c, d) (max_over_time(aggr +-+-+-+ | 0_| 0_|_SortPreservingMergeExec: [b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST] REDACTED |_|_|_SortExec: expr=[b@0 ASC NULLS LAST, c@1 ASC NULLS LAST, d@2 ASC NULLS LAST, greptime_timestamp@3 ASC NULLS LAST], preserve_partitioning=[true] REDACTED -|_|_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_merge(__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_merge(__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_AggregateExec: mode=Partial, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_merge(__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_merge(__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))) as min(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| -| 1_| 0_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +| 1_| 0_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_AggregateExec: mode=Partial, gby=[b@2 as b, c@3 as c, d@4 as d, greptime_timestamp@0 as greptime_timestamp], aggr=[__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[b@2 as b, c@3 as c, d@4 as d, greptime_timestamp@0 as greptime_timestamp], aggr=[__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_FilterExec: prom_max_over_time(greptime_timestamp_range,greptime_value)@1 IS NOT NULL REDACTED |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, prom_max_over_time(greptime_timestamp_range@6, greptime_value@5) as prom_max_over_time(greptime_timestamp_range,greptime_value), b@1 as b, c@2 as c, d@3 as d] REDACTED |_|_|_PromRangeManipulateExec: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp] REDACTED @@ -361,9 +363,9 @@ tql analyze (1752591864, 1752592164, '30s') min by (b, c, d) (max_over_time(aggr |_|_|_PromSeriesDivideExec: tags=["a", "b", "c", "d"] REDACTED |_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED |_|_|_| -| 1_| 1_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +| 1_| 1_|_AggregateExec: mode=FinalPartitioned, gby=[b@0 as b, c@1 as c, d@2 as d, greptime_timestamp@3 as greptime_timestamp], aggr=[__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_AggregateExec: mode=Partial, gby=[b@2 as b, c@3 as c, d@4 as d, greptime_timestamp@0 as greptime_timestamp], aggr=[__min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[b@2 as b, c@3 as c, d@4 as d, greptime_timestamp@0 as greptime_timestamp], aggr=[__prom_min_state(prom_max_over_time(greptime_timestamp_range,greptime_value))] REDACTED |_|_|_FilterExec: prom_max_over_time(greptime_timestamp_range,greptime_value)@1 IS NOT NULL REDACTED |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, prom_max_over_time(greptime_timestamp_range@6, greptime_value@5) as prom_max_over_time(greptime_timestamp_range,greptime_value), b@1 as b, c@2 as c, d@3 as d] REDACTED |_|_|_PromRangeManipulateExec: req range=[1752591864000..1752592164000], interval=[30000], eval range=[120000], time index=[greptime_timestamp] REDACTED diff --git a/tests/cases/standalone/common/promql/count_values.result b/tests/cases/standalone/common/promql/count_values.result index 886c3a143b7..1d04301d73b 100644 --- a/tests/cases/standalone/common/promql/count_values.result +++ b/tests/cases/standalone/common/promql/count_values.result @@ -84,7 +84,7 @@ TQL EVAL (0, 15, '5s') abs(count_values("v", http_requests)); TQL EVAL (0, 15, '5s') round(count_values("v", http_requests)); +---------------------+-------------------------------------------------+-----+ -| ts | prom_round(count(http_requests.val),Float64(0)) | v | +| ts | prom_round(count(http_requests.val),Float64(1)) | v | +---------------------+-------------------------------------------------+-----+ | 1970-01-01T00:00:00 | 1.0 | 401 | | 1970-01-01T00:00:00 | 3.0 | 200 | diff --git a/tests/cases/standalone/common/promql/delta_temporality.result b/tests/cases/standalone/common/promql/delta_temporality.result index 787e72e8dc8..ac32a907cb9 100644 --- a/tests/cases/standalone/common/promql/delta_temporality.result +++ b/tests/cases/standalone/common/promql/delta_temporality.result @@ -162,7 +162,7 @@ TQL EVAL (180, 180, '1m') round(sum(rate(delta_temporality[3m])), 0.000001); +---------------------+----------------------------------------------------------------------------------------+ | ts | prom_round(sum(prom_rate(ts_range,greptime_value,ts,Int64(180000))),Float64(0.000001)) | +---------------------+----------------------------------------------------------------------------------------+ -| 1970-01-01T00:03:00 | 0.45555599999999996 | +| 1970-01-01T00:03:00 | 0.455556 | +---------------------+----------------------------------------------------------------------------------------+ CREATE TABLE delta_marker_only ( diff --git a/tests/cases/standalone/common/promql/dotted_column_names.result b/tests/cases/standalone/common/promql/dotted_column_names.result index c855a5ba5c7..bb4912bc94e 100644 --- a/tests/cases/standalone/common/promql/dotted_column_names.result +++ b/tests/cases/standalone/common/promql/dotted_column_names.result @@ -424,9 +424,9 @@ TQL EXPLAIN (0, 10, '5s') topk(1, {"otel.m"}); | plan_type_| plan_| +-+-+ | logical_plan_| Projection: otel.m.v.val, otel.m.service.name, otel.m.Host, otel.m.host, otel.m.ts.time_| -|_|_Sort: otel.m.ts.time ASC NULLS LAST, row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ASC NULLS LAST_| -|_|_Filter: CAST(row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS Float64) <= Float64(1)_| -|_|_WindowAggr: windowExpr=[[row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]_| +|_|_Sort: otel.m.ts.time ASC NULLS LAST, row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ASC NULLS LAST_| +|_|_Filter: CAST(row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS Float64) <= Float64(1)_| +|_|_WindowAggr: windowExpr=[[row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]_| |_|_MergeScan [is_placeholder=false, remote_input=[_| |_| PromInstantManipulate: range=[0..10000], lookback=[300000], interval=[5000], time index=[ts.time]_| |_|_PromSeriesDivide: tags=["service.name", "Host", "host"]_| @@ -435,10 +435,10 @@ TQL EXPLAIN (0, 10, '5s') topk(1, {"otel.m"}); |_|_TableScan: otel.m, partial_filters=[otel.m.ts.time >= TimestampMillisecond(-299999, None), otel.m.ts.time <= TimestampMillisecond(10000, None)]_| |_| ]]_| | physical_plan | ProjectionExec: expr=[v.val@4 as v.val, service.name@1 as service.name, Host@2 as Host, host@3 as host, ts.time@0 as ts.time]_| -|_|_SortPreservingMergeExec: [ts.time@0 ASC NULLS LAST, row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 ASC NULLS LAST]_| -|_|_FilterExec: CAST(row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 AS Float64) <= 1_| -|_|_BoundedWindowAggExec: wdw=[row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [otel.m.ts.time] ORDER BY [otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] | -|_|_SortExec: expr=[ts.time@0 ASC NULLS LAST, v.val@4 DESC, service.name@1 DESC, Host@2 DESC, host@3 DESC], preserve_REDACTED +|_|_SortPreservingMergeExec: [ts.time@0 ASC NULLS LAST, row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 ASC NULLS LAST]_| +|_|_FilterExec: CAST(row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 AS Float64) <= 1_| +|_|_BoundedWindowAggExec: wdw=[row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [otel.m.ts.time] ORDER BY [isnan(otel.m.v.val) ASC NULLS FIRST, otel.m.v.val DESC NULLS FIRST, otel.m.service.name DESC NULLS FIRST, otel.m.Host DESC NULLS FIRST, otel.m.host DESC NULLS FIRST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] | +|_|_SortExec: expr=[ts.time@0 ASC NULLS LAST, isnan(v.val@4) ASC, v.val@4 DESC, service.name@1 DESC, Host@2 DESC, host@3 DESC], preserve_REDACTED |_|_RepartitionExec: REDACTED |_|_MergeScanExec: REDACTED |_|_| diff --git a/tests/cases/standalone/common/promql/float_semantics.result b/tests/cases/standalone/common/promql/float_semantics.result new file mode 100644 index 00000000000..e56db8b0ba6 --- /dev/null +++ b/tests/cases/standalone/common/promql/float_semantics.result @@ -0,0 +1,115 @@ +-- NaN and rounding follow Prometheus (IEEE 754) rather than SQL ordering. +-- Expected results are from Prometheus 3. +create table float_semantics ( + ts timestamp time index, + val double, + host string primary key +); + +Affected Rows: 0 + +insert into float_semantics values + (0, 1.0, 'a'), + (0, 2.0, 'b'), + (0, 'NaN'::double, 'c'), + (0, -2.5, 'd'); + +Affected Rows: 4 + +tql eval (0, 0, '1s') max(float_semantics); + ++---------------------+--------------------------+ +| ts | max(float_semantics.val) | ++---------------------+--------------------------+ +| 1970-01-01T00:00:00 | 2.0 | ++---------------------+--------------------------+ + +tql eval (0, 0, '1s') min(float_semantics); + ++---------------------+--------------------------+ +| ts | min(float_semantics.val) | ++---------------------+--------------------------+ +| 1970-01-01T00:00:00 | -2.5 | ++---------------------+--------------------------+ + +tql eval (0, 0, '1s') topk(1, float_semantics); + ++-----+------+---------------------+ +| val | host | ts | ++-----+------+---------------------+ +| 2.0 | b | 1970-01-01T00:00:00 | ++-----+------+---------------------+ + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') float_semantics <= NaN; + ++----+-----+------+ +| ts | val | host | ++----+-----+------+ ++----+-----+------+ + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') float_semantics != bool NaN; + ++------+---------------------+---------------------+ +| host | ts | val != Float64(NaN) | ++------+---------------------+---------------------+ +| a | 1970-01-01T00:00:00 | 1.0 | +| b | 1970-01-01T00:00:00 | 1.0 | +| c | 1970-01-01T00:00:00 | 1.0 | +| d | 1970-01-01T00:00:00 | 1.0 | ++------+---------------------+---------------------+ + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') round(float_semantics); + ++---------------------+----------------------------+------+ +| ts | prom_round(val,Float64(1)) | host | ++---------------------+----------------------------+------+ +| 1970-01-01T00:00:00 | -2.0 | d | +| 1970-01-01T00:00:00 | 1.0 | a | +| 1970-01-01T00:00:00 | 2.0 | b | +| 1970-01-01T00:00:00 | NaN | c | ++---------------------+----------------------------+------+ + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') sqrt(float_semantics); + ++---------------------+--------------------+------+ +| ts | sqrt(val) | host | ++---------------------+--------------------+------+ +| 1970-01-01T00:00:00 | 1.0 | a | +| 1970-01-01T00:00:00 | 1.4142135623730951 | b | +| 1970-01-01T00:00:00 | NaN | c | +| 1970-01-01T00:00:00 | NaN | d | ++---------------------+--------------------+------+ + +-- prom_min/prom_max have their own column names, so SQL can use them next to min/max. +select min(val) as sql_lo, prom_min(val) as prom_lo from float_semantics; + ++--------+---------+ +| sql_lo | prom_lo | ++--------+---------+ +| -2.5 | -2.5 | ++--------+---------+ + +select max(val) as sql_hi, prom_max(val) as prom_hi from float_semantics; + ++--------+---------+ +| sql_hi | prom_hi | ++--------+---------+ +| NaN | 2.0 | ++--------+---------+ + +select prom_min(val) as lo from float_semantics having min(val) < 0; + ++------+ +| lo | ++------+ +| -2.5 | ++------+ + +drop table float_semantics; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/promql/float_semantics.sql b/tests/cases/standalone/common/promql/float_semantics.sql new file mode 100644 index 00000000000..f23e95141c4 --- /dev/null +++ b/tests/cases/standalone/common/promql/float_semantics.sql @@ -0,0 +1,41 @@ +-- NaN and rounding follow Prometheus (IEEE 754) rather than SQL ordering. +-- Expected results are from Prometheus 3. + +create table float_semantics ( + ts timestamp time index, + val double, + host string primary key +); + +insert into float_semantics values + (0, 1.0, 'a'), + (0, 2.0, 'b'), + (0, 'NaN'::double, 'c'), + (0, -2.5, 'd'); + +tql eval (0, 0, '1s') max(float_semantics); + +tql eval (0, 0, '1s') min(float_semantics); + +tql eval (0, 0, '1s') topk(1, float_semantics); + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') float_semantics <= NaN; + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') float_semantics != bool NaN; + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') round(float_semantics); + +-- SQLNESS SORT_RESULT 3 1 +tql eval (0, 0, '1s') sqrt(float_semantics); + +-- prom_min/prom_max have their own column names, so SQL can use them next to min/max. +select min(val) as sql_lo, prom_min(val) as prom_lo from float_semantics; + +select max(val) as sql_hi, prom_max(val) as prom_hi from float_semantics; + +select prom_min(val) as lo from float_semantics having min(val) < 0; + +drop table float_semantics; diff --git a/tests/cases/standalone/common/promql/functions.result b/tests/cases/standalone/common/promql/functions.result index 325fbccd17d..1867413aec8 100644 --- a/tests/cases/standalone/common/promql/functions.result +++ b/tests/cases/standalone/common/promql/functions.result @@ -109,7 +109,7 @@ tql eval (4, 4, '1s') quantile_over_time(0.4 + 0.1, prom_series_q[4s]); tql eval (3, 3, '1s') round(prom_series_q); +---------------------+----------------------------+------+ -| ts | prom_round(val,Float64(0)) | host | +| ts | prom_round(val,Float64(1)) | host | +---------------------+----------------------------+------+ | 1970-01-01T00:00:03 | 346.0 | q | +---------------------+----------------------------+------+ @@ -133,8 +133,8 @@ tql eval (1, 4, '1s') round(prom_series_q, 0.05 + 0.05); | ts | prom_round(val,Float64(0.05) + Float64(0.05)) | host | +---------------------+-----------------------------------------------+------+ | 1970-01-01T00:00:01 | 123.5 | q | -| 1970-01-01T00:00:02 | 234.60000000000002 | q | -| 1970-01-01T00:00:03 | 345.70000000000005 | q | +| 1970-01-01T00:00:02 | 234.6 | q | +| 1970-01-01T00:00:03 | 345.7 | q | | 1970-01-01T00:00:04 | 456.8 | q | +---------------------+-----------------------------------------------+------+ @@ -156,10 +156,10 @@ tql eval (1, 4, '1s') round(prom_series_q, 100.0 + 3.0); +---------------------+-------------------------------------------+------+ | ts | prom_round(val,Float64(100) + Float64(3)) | host | +---------------------+-------------------------------------------+------+ -| 1970-01-01T00:00:01 | 103.0 | q | -| 1970-01-01T00:00:02 | 206.0 | q | +| 1970-01-01T00:00:01 | 103.00000000000001 | q | +| 1970-01-01T00:00:02 | 206.00000000000003 | q | | 1970-01-01T00:00:03 | 309.0 | q | -| 1970-01-01T00:00:04 | 412.0 | q | +| 1970-01-01T00:00:04 | 412.00000000000006 | q | +---------------------+-------------------------------------------+------+ -- SQLNESS SORT_RESULT 3 1 diff --git a/tests/cases/standalone/common/promql/tsid_binary_join_regression.result b/tests/cases/standalone/common/promql/tsid_binary_join_regression.result index fc4bc6f01c2..0288a045390 100644 --- a/tests/cases/standalone/common/promql/tsid_binary_join_regression.result +++ b/tests/cases/standalone/common/promql/tsid_binary_join_regression.result @@ -336,7 +336,7 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > tsid_binary_join_right; +-+-+-+ | stage | node | plan_| +-+-+-+ -| 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@3, __tsid@1), (ts@4, ts@2)], filter=greptime_value@1 < greptime_value@0, projection=[ts@4, greptime_value@0, host@1, job@2, __tsid@3], NullsEqual: true REDACTED +| 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@3, __tsid@1), (ts@4, ts@2)], filter=prom_gt(greptime_value@0, greptime_value@1), projection=[ts@4, greptime_value@0, host@1, job@2, __tsid@3], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([__tsid@3, ts@4],REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([__tsid@1, ts@2],REDACTED @@ -374,8 +374,8 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > bool tsid_binary_join_right; +-+-+-+ | stage | node | plan_| +-+-+-+ -| 0_| 0_|_ProjectionExec: expr=[host@0 as host, job@1 as job, ts@2 as ts, __tsid@3 as __tsid, CAST(greptime_value@4 < greptime_value@5 AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED -|_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@2, ts@4)], projection=[host@4, job@5, ts@7, __tsid@6, greptime_value@3, greptime_value@0], NullsEqual: true REDACTED +| 0_| 0_|_ProjectionExec: expr=[host@0 as host, job@1 as job, ts@2 as ts, __tsid@3 as __tsid, CAST(prom_gt(greptime_value@4, greptime_value@5) AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED +|_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@2, ts@4)], projection=[host@4, job@5, ts@7, __tsid@6, greptime_value@0, greptime_value@3], NullsEqual: true REDACTED |_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_CooperativeExec REDACTED |_|_|_MergeScanExec: REDACTED @@ -416,7 +416,7 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > tsid_binary_join_right) / tsi | 0_| 0_|_ProjectionExec: expr=[host@0 as host, job@1 as job, ts@2 as ts, __tsid@3 as __tsid, greptime_value@4 / greptime_value@5 * 100 as .greptime_value / tsid_binary_join_left.greptime_value * Float64(100)] REDACTED |_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@2, __tsid@3), (ts@0, ts@4)], projection=[host@4, job@5, ts@7, __tsid@6, greptime_value@1, greptime_value@3], NullsEqual: true REDACTED |_|_|_CoalescePartitionsExec REDACTED -|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], filter=greptime_value@1 < greptime_value@0, projection=[ts@2, greptime_value@0, __tsid@1], NullsEqual: true REDACTED +|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], filter=prom_gt(greptime_value@0, greptime_value@1), projection=[ts@2, greptime_value@0, __tsid@1], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED |_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED @@ -467,8 +467,8 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > bool tsid_binary_join_right) |_|_|_CoalescePartitionsExec REDACTED |_|_|_ProjectionExec: expr=[ts@0 as ts, __tsid@1 as __tsid, tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value@2 + greptime_value@3 as tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value] REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@0, ts@2)], projection=[ts@5, __tsid@4, tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value@2, greptime_value@3], NullsEqual: true REDACTED -|_|_|_ProjectionExec: expr=[ts@0 as ts, __tsid@1 as __tsid, CAST(greptime_value@2 < greptime_value@3 AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED -|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], projection=[ts@5, __tsid@4, greptime_value@3, greptime_value@0], NullsEqual: true REDACTED +|_|_|_ProjectionExec: expr=[ts@0 as ts, __tsid@1 as __tsid, CAST(prom_gt(greptime_value@2, greptime_value@3) AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED +|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], projection=[ts@5, __tsid@4, greptime_value@0, greptime_value@3], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED |_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED