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 <killme2008@gmail.com> * 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 <killme2008@gmail.com> * 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 <killme2008@gmail.com> * test(promql): cover NaN and rounding semantics Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * 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 <killme2008@gmail.com> * 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 <killme2008@gmail.com> * 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 <killme2008@gmail.com> * 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 <killme2008@gmail.com> * test: restore column names in sqlness results for PromQL float semantics Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * test(promql): share the prom_min/prom_max groups accumulator with tests Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix(promql): keep distinct names for prom_min/prom_max Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com> (cherry picked from commit f3a0841f4f) Signed-off-by: WenyXu <wenymedia@gmail.com>
2026-10-12 15:02:32 +00:00 · 2026-10-12 12:49:21 +00:00
parent 5fe2a57adb
commit 99ef93e143
18 changed files with 954 additions and 140 deletions
@@ -14,6 +14,7 @@

 pub mod approximate;
 pub mod count_hash;
+pub mod extremum;
 #[cfg(feature = "geo")]
 pub mod geo;
 pub mod vector;
@@ -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<DataType> {
+        Ok(DataType::Float64)
+    }
+
+    fn accumulator(&self, _acc_args: AccumulatorArgs) -> DfResult<Box<dyn Accumulator>> {
+        Ok(Box::new(ExtremumAccumulator {
+            is_max: self.is_max,
+            value: None,
+        }))
+    }
+
+    fn state_fields(&self, args: StateFieldsArgs) -> DfResult<Vec<FieldRef>> {
+        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<Box<dyn GroupsAccumulator>> {
+        Ok(groups_accumulator(self.is_max))
+    }
+}
+
+fn groups_accumulator(is_max: bool) -> Box<dyn GroupsAccumulator> {
+    Box::new(
+        PrimitiveGroupsAccumulator::<Float64Type, _>::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<f64>,
+}
+
+impl ExtremumAccumulator {
+    fn update(&mut self, values: &ArrayRef) {
+        for v in values.as_primitive::<Float64Type>().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<Vec<ScalarValue>> {
+        Ok(vec![ScalarValue::Float64(self.value)])
+    }
+
+    fn evaluate(&mut self) -> DfResult<ScalarValue> {
+        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<Option<f64>>,
+        groups: Vec<usize>,
+    ) -> Vec<Option<f64>> {
+        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::<Float64Type>().iter().collect()
+    }
+
+    fn is_nan(v: Option<f64>) -> 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)));
+    }
+}
@@ -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<Arc<FunctionRegistry>> = 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);

@@ -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,
@@ -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<ScalarUDF> {
+        [
+            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<DataType> {
+        Ok(DataType::Boolean)
+    }
+
+    fn schema_name(&self, args: &[Expr]) -> DfResult<String> {
+        Ok(format!(
+            "{} {} {}",
+            args[0].schema_name(),
+            self.op,
+            args[1].schema_name()
+        ))
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DfResult<ColumnarValue> {
+        let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+        let lhs = arrays[0].as_primitive::<Float64Type>();
+        let rhs = arrays[1].as_primitive::<Float64Type>();
+        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<DataType> {
+        Ok(DataType::Float64)
+    }
+
+    fn schema_name(&self, args: &[Expr]) -> DfResult<String> {
+        Ok(format!("sqrt({})", args[0].schema_name()))
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DfResult<ColumnarValue> {
+        let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+        let input = arrays[0].as_primitive::<Float64Type>();
+        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<ArrayRef>, 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::<Float64Type>();
+        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)");
+    }
+}
@@ -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<Float64Type> =
                    value_array.as_primitive::<Float64Type>().unary(op);
                Ok(ColumnarValue::Array(Arc::new(result) as _))
@@ -90,14 +84,10 @@ impl Round {
                )?;

                let result: PrimitiveArray<Float64Type> =
-                    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::<Float64Type>().value(0).is_nan());
+    }
 }
@@ -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"
        )
    }

@@ -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::<Vec<_>>();

-        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<Box<PromExpr>>,
@@ -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!(
@@ -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]\
@@ -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::<false>::scalar_udf()));
        let _ = session_state.register_udf(Arc::new(IDelta::<true>::scalar_udf()));
@@ -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
@@ -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 |
@@ -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 (
@@ -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
 |_|_|
@@ -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
+
@@ -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;
@@ -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
@@ -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