From 7c139a2ccca20ba62960f6a97b305e033ee347a8 Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Thu, 22 Oct 2026 02:31:40 +0800 Subject: [PATCH] feat(promql): skip native histogram samples in math functions and cover review gaps Signed-off-by: Dennis Zhuang --- src/promql/src/map.rs | 71 +++++++++--- src/promql/src/map/broadcast.rs | 91 ++++++++------- src/promql/src/map/scalar_parameter_cases.txt | 55 +++++---- src/promql/src/map/tests.rs | 51 +++++--- src/promql/src/map/tests/broadcast.rs | 109 ++++++++++++++---- 5 files changed, 252 insertions(+), 125 deletions(-) diff --git a/src/promql/src/map.rs b/src/promql/src/map.rs index a23e1cbe62f..0e2ff67fe02 100644 --- a/src/promql/src/map.rs +++ b/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> { 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) } diff --git a/src/promql/src/map/broadcast.rs b/src/promql/src/map/broadcast.rs index d2f027852dc..3ddf82edba7 100644 --- a/src/promql/src/map/broadcast.rs +++ b/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, series: &Series, - function: ParameterizedMath, - parameters: &[&Series], + function: ParameterizedMath<'_>, ) -> Result> { - 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 ¶meters { 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, }); } diff --git a/src/promql/src/map/scalar_parameter_cases.txt b/src/promql/src/map/scalar_parameter_cases.txt index 9415f983921..5fe9df9dc82 100644 --- a/src/promql/src/map/scalar_parameter_cases.txt +++ b/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 diff --git a/src/promql/src/map/tests.rs b/src/promql/src/map/tests.rs index 1cc936287fd..66116e4d916 100644 --- a/src/promql/src/map/tests.rs +++ b/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 { } #[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] diff --git a/src/promql/src/map/tests/broadcast.rs b/src/promql/src/map/tests/broadcast.rs index 62d6b674155..7d8022ef967 100644 --- a/src/promql/src/map/tests/broadcast.rs +++ b/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, 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, ¶meters).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, &[¶meter]) - .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)])]); }