feat(promql): skip native histogram samples in math functions and cover review gaps

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

src/promql/src/map.rs

@@ -42,6 +42,7 @@ use datatypes::series::samples::{
 use datatypes::series::time::{GridRange, TimeAxis};
 
 use crate::functions::Round;
+use crate::grid_series::GridSeriesBuilder;
 
 /// Float math supported by the evaluated-series kernel.
 #[derive(Clone, Copy, Debug)]
@@ -117,9 +118,12 @@ pub enum ScalarOperand {
 ///
 /// Eval must resolve raw staleness before calling this kernel. Evaluated samples use validity,
 /// not NaN payloads, for missingness: arithmetic can produce the reserved stale bit pattern too.
-/// Raw time axes and histogram/mixed segments are rejected. Unsupported query types must be
-/// admitted to the ordinary path before execution; this function neither falls back nor returns
-/// partial results.
+/// Raw time axes are rejected.
+///
+/// Native histogram samples are skipped, as Prometheus' math functions process only float
+/// samples: a point holding a histogram has no result in that field, and a series with no float
+/// sample left has none at all. Float-only fields keep their chunks; a field with histograms is
+/// rebuilt point by point.
 ///
 /// Returns `None` for `clamp` with `max < min`, which yields no series in Prometheus.
 ///
@@ -128,9 +132,31 @@ pub enum ScalarOperand {
 /// `mergeSeriesWithSameLabelset`).
 pub fn map_math(schema: &SeriesSchema, series: &Series, op: MathOp) -> Result<Option<Series>> {
     if op.is_empty_clamp() {
-        validate_float_series(schema, series)?;
+        validate_grid_series(schema, series)?;
         return Ok(None);
     }
+    if !is_float_series(series) {
+        let time = validate_grid_series(schema, series)?;
+        let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone()))?;
+        let mut cursors: Vec<_> = series.fields.iter().map(|field| field.cursor()).collect();
+        let mut row = Vec::with_capacity(cursors.len());
+        for (point, index) in time.positions() {
+            row.clear();
+            for cursor in &mut cursors {
+                row.push(
+                    match cursor
+                        .at(index)
+                        .expect("the axis and the fields are aligned")
+                    {
+                        SampleRef::Float(value) => Some(op.calculate(value)),
+                        SampleRef::Missing | SampleRef::Histogram(_) => None,
+                    },
+                );
+            }
+            builder.push(point, &row)?;
+        }
+        return builder.finish(drop_metric_name(&series.metric)?);
+    }
     let mut output = map_float_fields(schema, series, |floats| op.apply(floats))?;
     output.metric = drop_metric_name(&output.metric)?;
     Ok(Some(output))
@@ -372,6 +398,31 @@ fn series_error(error: datatypes::error::Error) -> DataFusionError {
 pub(crate) fn validate_float_series<'a>(
     schema: &SeriesSchema,
     series: &'a Series,
+) -> Result<&'a TimeAxis> {
+    let time = validate_grid_series(schema, series)?;
+    if !is_float_series(series) {
+        return Err(DataFusionError::NotImplemented(
+            "Series operators support only float samples; histogram/mixed samples are not supported"
+                .into(),
+        ));
+    }
+    Ok(time)
+}
+
+/// Whether every field of `series` holds float samples only.
+fn is_float_series(series: &Series) -> bool {
+    series.fields.iter().all(|field| {
+        field
+            .segments()
+            .iter()
+            .all(|segment| matches!(segment.dense(), Some(DenseSamplesRef::Float(_))))
+    })
+}
+
+/// Checks `series` against `schema` and its evaluation grid, whatever its sample types.
+pub(crate) fn validate_grid_series<'a>(
+    schema: &SeriesSchema,
+    series: &'a Series,
 ) -> Result<&'a TimeAxis> {
     let time = &series.time;
     if time.grid().is_none() {
@@ -386,18 +437,6 @@ pub(crate) fn validate_float_series<'a>(
         series.fields.clone(),
     )
     .map_err(series_error)?;
-    for field in series.fields.iter() {
-        if field
-            .segments()
-            .iter()
-            .any(|segment| !matches!(segment.dense(), Some(DenseSamplesRef::Float(_))))
-        {
-            return Err(DataFusionError::NotImplemented(
-                "Series operators support only float samples; histogram/mixed samples are not supported"
-                    .into(),
-            ));
-        }
-    }
     Ok(time)
 }
 

src/promql/src/map/broadcast.rs

@@ -16,12 +16,13 @@ use std::sync::Arc;
 
 use datafusion::error::{DataFusionError, Result};
 use datatypes::series::MetricName;
-use datatypes::series::samples::{FieldSpec, Series, SeriesSchema};
+use datatypes::series::samples::{FieldSpec, SampleRef, Series, SeriesSchema};
 use datatypes::series::time::TimeAxis;
 
 use crate::grid_series::GridSeriesBuilder;
 use crate::map::{
     ArithmeticOp, Comparison, MathOp, drop_metric_name, float_sample, validate_float_series,
+    validate_grid_series,
 };
 
 /// A time-varying, unlabelled single-field scalar, with explicit operand direction.
@@ -129,62 +130,60 @@ pub fn map_scalar_function(
 }
 
 /// A math function whose parameters are scalar expressions, which Prometheus evaluates at every
-/// step: `round(v, to_nearest)`, `clamp(v, min, max)`, `clamp_min(v, min)` and `clamp_max(v, max)`.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum ParameterizedMath {
-    Round,
-    Clamp,
-    ClampMin,
-    ClampMax,
+/// step, with those parameters as time-varying scalars.
+#[derive(Clone, Copy, Debug)]
+pub enum ParameterizedMath<'a> {
+    /// `round(v, to_nearest)`.
+    Round { to_nearest: &'a Series },
+    /// `clamp(v, min, max)`.
+    Clamp { min: &'a Series, max: &'a Series },
+    /// `clamp_min(v, min)`.
+    ClampMin { min: &'a Series },
+    /// `clamp_max(v, max)`.
+    ClampMax { max: &'a Series },
 }
 
-impl ParameterizedMath {
-    /// The function with the parameters of one point, in argument order.
-    fn at(self, parameters: &[f64]) -> MathOp {
+impl<'a> ParameterizedMath<'a> {
+    /// The parameters in argument order.
+    fn parameters(self) -> Vec<&'a Series> {
         match self {
-            Self::Round => MathOp::Round {
-                nearest: parameters[0],
-            },
-            Self::Clamp => MathOp::Clamp {
-                min: parameters[0],
-                max: parameters[1],
-            },
-            Self::ClampMin => MathOp::ClampMin { min: parameters[0] },
-            Self::ClampMax => MathOp::ClampMax { max: parameters[0] },
+            Self::Round { to_nearest } => vec![to_nearest],
+            Self::Clamp { min, max } => vec![min, max],
+            Self::ClampMin { min } => vec![min],
+            Self::ClampMax { max } => vec![max],
         }
     }
 
-    fn arity(self) -> usize {
+    /// The function with the parameter values of one point, in argument order.
+    fn at(self, values: &[f64]) -> MathOp {
         match self {
-            Self::Clamp => 2,
-            Self::Round | Self::ClampMin | Self::ClampMax => 1,
+            Self::Round { .. } => MathOp::Round { nearest: values[0] },
+            Self::Clamp { .. } => MathOp::Clamp {
+                min: values[0],
+                max: values[1],
+            },
+            Self::ClampMin { .. } => MathOp::ClampMin { min: values[0] },
+            Self::ClampMax { .. } => MathOp::ClampMax { max: values[0] },
         }
     }
 }
 
-/// Applies `function` to every float field with its parameters read at each point from
-/// time-varying scalars, and drops the metric name, as [`crate::map::map_math`] does with
-/// constant parameters.
+/// Applies `function` to every field with its parameters read at each point from time-varying
+/// scalars, and drops the metric name, as [`crate::map::map_math`] does with constant parameters;
+/// native histogram samples are skipped the same way.
 ///
 /// A point where a parameter has no value has no result. `clamp` with `max < min` at a point has
 /// no result there, as Prometheus returns an empty vector for that step. `None` means no point
-/// holds a result.
+/// holds a result. Same-labelset merging is left to the caller.
 pub fn map_math_with_parameters(
     schema: &Arc<SeriesSchema>,
     series: &Series,
-    function: ParameterizedMath,
-    parameters: &[&Series],
+    function: ParameterizedMath<'_>,
 ) -> Result<Option<Series>> {
-    if parameters.len() != function.arity() {
-        return Err(DataFusionError::Execution(format!(
-            "{function:?} takes {} scalar parameters, got {}",
-            function.arity(),
-            parameters.len()
-        )));
-    }
-    let time = validate_float_series(schema, series)?;
+    let time = validate_grid_series(schema, series)?;
+    let parameters = function.parameters();
     let mut parameter_axes = Vec::with_capacity(parameters.len());
-    for parameter in parameters {
+    for parameter in &parameters {
         parameter_axes.push(validate_scalar(schema, parameter)?);
     }
     // Points only increase, so each parameter's row does too and one cursor reads it.
@@ -204,16 +203,16 @@ pub fn map_math_with_parameters(
             });
             values.extend(value);
         }
+        let op = (values.len() == parameters.len())
+            .then(|| function.at(&values))
+            .filter(|op| !op.is_empty_clamp());
         row.clear();
-        let op = (values.len() == parameters.len()).then(|| function.at(&values));
         for cursor in &mut field_cursors {
-            let sample = float_sample(
-                cursor
-                    .at(index)
-                    .expect("the axis and the fields are aligned"),
-            );
-            row.push(match op {
-                Some(op) if !op.is_empty_clamp() => sample.map(|value| op.calculate(value)),
+            let sample = cursor
+                .at(index)
+                .expect("the axis and the fields are aligned");
+            row.push(match (op, sample) {
+                (Some(op), SampleRef::Float(value)) => Some(op.calculate(value)),
                 _ => None,
             });
         }

src/promql/src/map/scalar_parameter_cases.txt

@@ -5,8 +5,8 @@ case round
 expr round(a, scalar(p0))
 grid 100000 10000 6
 series __name__=a,i=0 ; 0:405edd2f1a9fbe77 1:c004000000000000 4:7ff8000000000001 5:3ff4000000000000
-param __name__=p0 ; 0:3fb999999999999a 1:4008000000000000 2:7ff0000000000000 3:3ff0000000000000 4:3fb999999999999a 5:3fb999999999999a
-result i=0 ; 0:405ee00000000000 1:c008000000000000 4:7ff8000000000001 5:3ff4cccccccccccd
+param __name__=p0 ; 0:fff0000000000000 1:7ff0000000000000 2:4008000000000000 3:3fe0000000000000 4:c000000000000000 5:7ff8000000000001
+result i=0 ; 0:fff8000000000000 1:fff8000000000000 4:7ff8000000000001 5:7ff8000000000001
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -45,7 +45,7 @@ expr round(a, scalar(p0))
 grid 100000 10000 1
 series __name__=a,i=0 ;
 series __name__=a,i=1 ;
-param __name__=p0 ; 0:4024000000000000
+param __name__=p0 ; 0:c000000000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -76,8 +76,8 @@ case round
 expr round(a, scalar(p0))
 grid 100000 10000 2
 series __name__=a,i=0 ; 0:405edd2f1a9fbe77
-param __name__=p0 ; 0:7ff8000000000001 1:4024000000000000
-result i=0 ; 0:7ff8000000000001
+param __name__=p0 ; 0:3ff0000000000000 1:fff0000000000000
+result i=0 ; 0:405ec00000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -114,8 +114,8 @@ case round
 expr round(a, scalar(p0))
 grid 100000 10000 1
 series __name__=a,i=0 ; 0:405edd2f1a9fbe77
-param __name__=p0 ; 0:c000000000000000
-result i=0 ; 0:405f000000000000
+param __name__=p0 ; 0:3fe0000000000000
+result i=0 ; 0:405ee00000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -150,9 +150,9 @@ expr round(a, scalar(p0))
 grid 100000 10000 4
 series __name__=a,i=0 ; 0:3ff4000000000000 1:7ff8000000000001 2:7ff0000000000000 3:7ff8000000000001
 series __name__=a,i=1 ; 1:7ff0000000000000 2:3ff4000000000000
-param __name__=p0 ; 0:4024000000000000 1:4008000000000000 2:0 3:4008000000000000
-result i=0 ; 0:0 1:7ff8000000000001 2:fff8000000000000 3:7ff8000000000001
-result i=1 ; 1:7ff0000000000000 2:fff8000000000000
+param __name__=p0 ; 0:7ff0000000000000 1:fff0000000000000 2:3fe0000000000000 3:3fb999999999999a
+result i=0 ; 0:fff8000000000000 1:7ff8000000000001 2:7ff0000000000000 3:7ff8000000000001
+result i=1 ; 1:fff8000000000000 2:3ff8000000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -181,8 +181,8 @@ expr round(a, scalar(p0))
 grid 100000 10000 1
 series __name__=a,i=0 ; 0:3e112e0be826d695
 series __name__=a,i=1 ;
-param __name__=p0 ; 0:7ff0000000000000
-result i=0 ; 0:fff8000000000000
+param __name__=p0 ; 0:7ff8000000000001
+result i=0 ; 0:7ff8000000000001
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -213,9 +213,9 @@ expr round(a, scalar(p0))
 grid 100000 10000 3
 series __name__=a,i=0 ; 2:0
 series __name__=a,i=1 ; 0:7ff0000000000000 2:401c000000000000
-param __name__=p0 ; 0:4024000000000000 1:7ff8000000000001 2:0
-result i=0 ; 2:fff8000000000000
-result i=1 ; 0:7ff0000000000000 2:fff8000000000000
+param __name__=p0 ; 0:4008000000000000 1:fff0000000000000 2:4008000000000000
+result i=0 ; 2:0
+result i=1 ; 0:7ff0000000000000 2:4018000000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -246,9 +246,9 @@ expr round(a, scalar(p0))
 grid 100000 10000 4
 series __name__=a,i=0 ; 1:8000000000000000 3:405edd2f1a9fbe77
 series __name__=a,i=1 ; 0:4004000000000000 1:fff0000000000000 2:3ff4000000000000 3:3ff4000000000000
-param __name__=p0 ; 0:7ff8000000000001 1:c000000000000000 2:3ff0000000000000 3:3fe0000000000000
-result i=0 ; 1:8000000000000000 3:405ee00000000000
-result i=1 ; 0:7ff8000000000001 1:fff0000000000000 2:3ff0000000000000 3:3ff8000000000000
+param __name__=p0 ; 0:7ff0000000000000 1:3fb999999999999a 2:fff0000000000000 3:c000000000000000
+result i=0 ; 1:0 3:405f000000000000
+result i=1 ; 0:fff8000000000000 1:fff0000000000000 2:fff8000000000000 3:4000000000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -283,8 +283,8 @@ expr round(a, scalar(p0))
 grid 100000 10000 3
 series __name__=a,i=0 ; 0:fff0000000000000 1:3e112e0be826d695 2:401c000000000000
 series __name__=a,i=1 ; 0:8000000000000000 2:fff0000000000000
-param __name__=p0 ; 0:4008000000000000 1:7ff8000000000001 2:4024000000000000
-result i=0 ; 0:fff0000000000000 1:7ff8000000000001 2:4024000000000000
+param __name__=p0 ; 0:3fe0000000000000 1:3fe0000000000000 2:4024000000000000
+result i=0 ; 0:fff0000000000000 1:0 2:4024000000000000
 result i=1 ; 0:0 2:fff0000000000000
 end
 case clamp
@@ -323,10 +323,10 @@ grid 100000 10000 5
 series __name__=a,i=0 ; 0:4004000000000000 1:7ff0000000000000
 series __name__=a,i=1 ; 0:c004000000000000 1:405edd2f1a9fbe77 3:8000000000000000
 series __name__=a,i=2 ; 0:0 1:3e112e0be826d695 2:3e112e0be826d695 3:405edd2f1a9fbe77
-param __name__=p0 ; 0:4008000000000000 1:7ff0000000000000 2:3fb999999999999a 3:3ff0000000000000 4:7ff0000000000000
-result i=0 ; 0:4008000000000000 1:fff8000000000000
-result i=1 ; 0:c008000000000000 1:fff8000000000000 3:0
-result i=2 ; 0:0 1:fff8000000000000 2:0 3:405ec00000000000
+param __name__=p0 ; 0:7ff0000000000000 1:4008000000000000 2:4024000000000000 3:0 4:c000000000000000
+result i=0 ; 0:fff8000000000000 1:7ff0000000000000
+result i=1 ; 0:fff8000000000000 1:405ec00000000000 3:fff8000000000000
+result i=2 ; 0:fff8000000000000 1:0 2:0 3:fff8000000000000
 end
 case clamp
 expr clamp(a, scalar(p0), scalar(p1))
@@ -356,3 +356,10 @@ series __name__=a,i=0 ; 0:3e112e0be826d695 1:3e112e0be826d695
 param __name__=p0 ; 0:4000000000000000 1:fff0000000000000 2:4000000000000000
 result i=0 ; 0:3e112e0be826d695 1:fff0000000000000
 end
+case round
+expr round(a, scalar(p0))
+grid 100000 10000 4
+series __name__=a,i=ties ; 0:4004000000000000 1:c004000000000000 2:3fe0000000000000 3:bfe0000000000000
+param __name__=p0 ; 0:3ff0000000000000 1:3ff0000000000000 2:3ff0000000000000 3:3ff0000000000000
+result i=ties ; 0:4008000000000000 1:c000000000000000 2:3ff0000000000000 3:0
+end

src/promql/src/map/tests.rs

@@ -31,7 +31,7 @@ use datatypes::series::histogram::{
     BucketSideView, CounterResetHint, HistogramBatch, HistogramBatchBuilder, HistogramView,
 };
 use datatypes::series::samples::{FieldSpec, HistogramSamples, SampleRef};
-use datatypes::series::time::{EvalGrid, GridRange, GridSpec};
+use datatypes::series::time::{EvalGrid, GridRange, GridSpan, GridSpec};
 use datatypes::series::{Label, MetricName};
 
 use super::*;
@@ -383,31 +383,50 @@ fn histogram_batch() -> Arc<HistogramBatch> {
 }
 
 #[test]
-fn histogram_and_mixed_fields_are_rejected_without_partial_success() {
+fn histogram_samples_are_skipped_and_float_samples_mapped() {
     let batch = histogram_batch();
     let histogram = Samples::try_from_histogram(UInt32Array::from(vec![0]), batch.clone()).unwrap();
     let mixed = Samples::try_from_mixed(
-        Float64Array::from(vec![Some(1.0), None]),
+        Float64Array::from(vec![Some(-1.0), None]),
         HistogramSamples::try_new(UInt32Array::from(vec![None, Some(0)]), batch.clone()).unwrap(),
     )
     .unwrap();
     let null_histogram = Samples::try_from_histogram(UInt32Array::from(vec![None]), batch).unwrap();
     let chunks = Samples::try_from_chunks(vec![floats(&[Some(-1.0)]), histogram.clone()]).unwrap();
-    let schema = schema(&["float", "unsupported"]);
-    for unsupported in [histogram, mixed, null_histogram, chunks] {
-        let prefix = floats(&vec![Some(-2.0); unsupported.len()]);
-        let input = series(&schema, 2, vec![prefix, unsupported]);
-        for op in [MathOp::Abs, MathOp::Clamp { min: 2.0, max: 1.0 }] {
-            assert!(matches!(
-                map_math(&schema, &input, op),
-                Err(DataFusionError::NotImplemented(_))
-            ));
-        }
-        assert_floats(
-            input.fields.get(0).unwrap(),
-            &vec![Some(-2.0); input.time.len()],
+    let two_fields = schema(&["float", "histograms"]);
+    // Prometheus' math functions process only float samples: each histogram is a point without a
+    // result, and the floats beside it, in its field or the other, are mapped as usual.
+    for (samples, expected) in [
+        (histogram.clone(), vec![None]),
+        (mixed, vec![Some(1.0), None]),
+        (null_histogram, vec![None]),
+        (chunks, vec![Some(1.0), None]),
+    ] {
+        let len = samples.len();
+        let input = series(
+            &two_fields,
+            2,
+            vec![floats(&vec![Some(-2.0); len]), samples],
+        );
+        let output = map_math(&two_fields, &input, MathOp::Abs).unwrap().unwrap();
+        assert_eq!(
+            run_floats(&output, 0),
+            vec![((2, len), vec![Some(2.0); len])]
+        );
+        assert_eq!(run_floats(&output, 1), vec![((2, len), expected)]);
+        assert!(
+            map_math(&two_fields, &input, MathOp::Clamp { min: 2.0, max: 1.0 })
+                .unwrap()
+                .is_none()
         );
     }
+    let only_histograms = schema(&["h"]);
+    let input = series(&only_histograms, 2, vec![histogram]);
+    assert!(
+        map_math(&only_histograms, &input, MathOp::Abs)
+            .unwrap()
+            .is_none()
+    );
 }
 
 #[test]

src/promql/src/map/tests/broadcast.rs

@@ -468,13 +468,7 @@ fn parameterized_math_matches_prometheus() {
         .filter(|case| case.contains("case "))
     {
         let lines: Vec<&str> = case.lines().filter(|line| !line.starts_with('#')).collect();
-        let function = match lines[0].strip_prefix("case ").unwrap() {
-            "round" => ParameterizedMath::Round,
-            "clamp" => ParameterizedMath::Clamp,
-            "clamp_min" => ParameterizedMath::ClampMin,
-            "clamp_max" => ParameterizedMath::ClampMax,
-            other => panic!("unknown function {other}"),
-        };
+        let name = lines[0].strip_prefix("case ").unwrap();
         let points: u64 = lines[2].split_whitespace().nth(3).unwrap().parse().unwrap();
         let grid = Arc::new(
             EvalGrid::try_new(GridSpec::try_new(100_000, 10_000, points).unwrap()).unwrap(),
@@ -486,18 +480,17 @@ fn parameterized_math_matches_prometheus() {
             .into(),
             eval_grid: Some(grid.clone()),
         });
-        let dense = |schema: &Arc<SeriesSchema>, labels: &Labels, values: &Points| {
-            crate::grid_series::reference::sparse_series(schema, labels, values)
-        };
+        let sparse_series = crate::grid_series::reference::sparse_series;
         let (mut inputs, mut parameters, mut expected) = (Vec::new(), Vec::new(), BTreeMap::new());
         for line in &lines[3..] {
             let (kind, rest) = line.split_once(' ').unwrap();
             let (labels, values) = parse(rest);
             match kind {
-                "series" => inputs.extend(dense(&input_schema, &labels, &values)),
+                "series" => inputs.extend(sparse_series(&input_schema, &labels, &values)),
                 // A parameter is `scalar(p)`: unlabelled, with a value at every point.
                 "param" => parameters.push(
-                    dense(&input_schema, &Labels::new(), &values).expect("a value at every point"),
+                    sparse_series(&input_schema, &Labels::new(), &values)
+                        .expect("a value at every point"),
                 ),
                 "result" => {
                     expected.insert(labels, values);
@@ -505,11 +498,16 @@ fn parameterized_math_matches_prometheus() {
                 other => panic!("unexpected line {other}"),
             }
         }
-        let parameters: Vec<&Series> = parameters.iter().collect();
+        let function = match (name, parameters.as_slice()) {
+            ("round", [to_nearest]) => ParameterizedMath::Round { to_nearest },
+            ("clamp", [min, max]) => ParameterizedMath::Clamp { min, max },
+            ("clamp_min", [min]) => ParameterizedMath::ClampMin { min },
+            ("clamp_max", [max]) => ParameterizedMath::ClampMax { max },
+            other => panic!("unknown case {other:?}"),
+        };
         let mut actual = BTreeMap::new();
         for input in &inputs {
-            if let Some(output) =
-                map_math_with_parameters(&input_schema, input, function, &parameters).unwrap()
+            if let Some(output) = map_math_with_parameters(&input_schema, input, function).unwrap()
             {
                 actual.insert(labels_of(&output.metric), points_of(&output));
             }
@@ -538,14 +536,79 @@ fn parameterized_math_matches_prometheus() {
     assert!(checked >= 40, "{checked}");
 }
 
-/// The parameter count must match the function.
+/// Parameters are read at each point of the input, whether they cover the grid or not: a point
+/// where a parameter has no value, absent from its axis or null, has no result, and every field
+/// takes the parameters of its point.
 #[test]
-fn parameterized_math_checks_its_parameter_count() {
-    let schema = Arc::new(schema(&["value"]));
-    let series = series(&schema, 0, vec![floats(&[Some(1.0)])]);
-    let parameter = scalar(&schema, 0, floats(&[Some(1.0)]));
-    assert!(
-        map_math_with_parameters(&schema, &series, ParameterizedMath::Clamp, &[&parameter])
-            .is_err()
+fn parameters_are_read_per_point_and_missing_ones_drop_the_point() {
+    let schema = Arc::new(schema(&["a", "b"]));
+    let input = series(
+        &schema,
+        2,
+        vec![
+            floats(&[Some(1.4), Some(2.6), Some(3.5), Some(4.4)]),
+            floats(&[Some(-1.4), None, Some(-3.5), Some(9.0)]),
+        ],
     );
+    // `min` covers 2..=3 and 5 with a null at 3; `max` covers 2..=5.
+    let min = Series::try_new(
+        &SeriesSchema {
+            fields: vec![FieldSpec {
+                name: "scalar".into(),
+            }]
+            .into(),
+            eval_grid: schema.eval_grid.clone(),
+        },
+        MetricName::try_new(None, vec![]).unwrap(),
+        TimeAxis::from_grid_spans(
+            schema.eval_grid.clone().unwrap(),
+            vec![GridSpan::new(2, 2), GridSpan::new(5, 1)],
+        )
+        .unwrap(),
+        SeriesFields::one(floats(&[Some(0.0), None, Some(5.0)])),
+    )
+    .unwrap();
+    let max = scalar(
+        &schema,
+        2,
+        floats(&[Some(2.0), Some(2.0), Some(2.0), Some(4.0)]),
+    );
+    let output = map_math_with_parameters(
+        &schema,
+        &input,
+        ParameterizedMath::Clamp {
+            min: &min,
+            max: &max,
+        },
+    )
+    .unwrap()
+    .unwrap();
+    // Point 2 clamps both fields to [0, 2]; point 3 has a null `min`; point 4 has no `min`; point 5
+    // has `max < min`, so no result.
+    assert_eq!(run_floats(&output, 0), vec![((2, 1), vec![Some(1.4)])]);
+    assert_eq!(run_floats(&output, 1), vec![((2, 1), vec![Some(0.0)])]);
+}
+
+/// Like `map_math`, the parameterized functions skip native histogram samples.
+#[test]
+fn parameterized_math_skips_histogram_samples() {
+    let schema = Arc::new(schema(&["mixed"]));
+    let mixed = Samples::try_from_mixed(
+        Float64Array::from(vec![Some(1.26), None]),
+        HistogramSamples::try_new(UInt32Array::from(vec![None, Some(0)]), histogram_batch())
+            .unwrap(),
+    )
+    .unwrap();
+    let input = series(&schema, 2, vec![mixed]);
+    let to_nearest = scalar(&schema, 2, floats(&[Some(0.5), Some(0.5)]));
+    let output = map_math_with_parameters(
+        &schema,
+        &input,
+        ParameterizedMath::Round {
+            to_nearest: &to_nearest,
+        },
+    )
+    .unwrap()
+    .unwrap();
+    assert_eq!(run_floats(&output, 0), vec![((2, 1), vec![Some(1.5)])]);
 }