chore: merge feat/narthex-window-eval-rate into feat/narthex-window-eval-overtime

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

src/promql/src/range_eval.rs

@@ -43,7 +43,7 @@ pub struct RangeWindow {
 }
 
 /// The valid samples of one field inside one window, in source order.
-pub struct FloatWindow<'a> {
+pub(crate) struct FloatWindow<'a> {
     /// Sample timestamps shifted by the offset onto the evaluation timeline; non-empty.
     pub timestamps: &'a [i64],
     pub values: &'a [f64],
@@ -53,7 +53,7 @@ pub struct FloatWindow<'a> {
 }
 
 /// Computes one window result; `None` means the window has no result.
-pub type WindowKernel = Box<dyn FnMut(FloatWindow<'_>) -> Option<f64> + Send>;
+pub(crate) type WindowKernel = Box<dyn FnMut(FloatWindow<'_>) -> Option<f64> + Send>;
 
 /// Evaluates range windows over raw float series, one series at a time.
 ///
@@ -89,8 +89,17 @@ struct FieldSamples {
 }
 
 impl RangeEvaluator {
-    /// Creates an evaluator for raw series of `input` onto `grid`.
+    /// Creates an evaluator of `function` for raw series of `input` onto `grid`.
     pub fn try_new(
+        input: &SeriesSchema,
+        grid: Arc<EvalGrid>,
+        window: RangeWindow,
+        function: RangeFunction,
+    ) -> Result<Self> {
+        Self::with_kernel(input, grid, window, function.kernel())
+    }
+
+    pub(crate) fn with_kernel(
         input: &SeriesSchema,
         grid: Arc<EvalGrid>,
         window: RangeWindow,

src/promql/src/range_eval/function.rs

@@ -46,7 +46,7 @@ pub enum RangeFunction {
 
 impl RangeFunction {
     /// Returns the kernel computing this function over one window.
-    pub fn kernel(self) -> WindowKernel {
+    pub(crate) fn kernel(self) -> WindowKernel {
         match self {
             Self::Rate => Box::new(extrapolated::<true, true>),
             Self::Increase => Box::new(extrapolated::<true, false>),

src/promql/src/range_eval/tests.rs

@@ -97,7 +97,7 @@ fn metric(host: u64) -> MetricName {
 
 fn probe_evaluator(fields: usize, grid: Arc<EvalGrid>, window: RangeWindow) -> RangeEvaluator {
     let kernel: WindowKernel = Box::new(probe);
-    RangeEvaluator::try_new(&raw_schema(fields), grid, window, kernel).unwrap()
+    RangeEvaluator::with_kernel(&raw_schema(fields), grid, window, kernel).unwrap()
 }
 
 /// Builds a chunk whose timestamp and value arrays are slices at a nonzero offset.
@@ -442,13 +442,9 @@ fn functions_match_range_udfs() {
             let mut rng = StdRng::seed_from_u64(seed);
             let fields = rng.random_range(1..=3);
             let (grid, window) = random_window(&mut rng);
-            let mut evaluator = RangeEvaluator::try_new(
-                &raw_schema(fields),
-                grid.clone(),
-                window,
-                function.kernel(),
-            )
-            .unwrap();
+            let mut evaluator =
+                RangeEvaluator::try_new(&raw_schema(fields), grid.clone(), window, function)
+                    .unwrap();
             for host in 0..3 {
                 let rows = random_rows(&mut rng, fields);
                 let expected = udf_reference(function, &rows, &grid, window, fields);
@@ -470,7 +466,7 @@ fn overlapping_window_spans_chunks() {
     };
     let kernel: WindowKernel = Box::new(sum);
     let mut evaluator =
-        RangeEvaluator::try_new(&raw_schema(1), grid(10, 10, 3), window, kernel).unwrap();
+        RangeEvaluator::with_kernel(&raw_schema(1), grid(10, 10, 3), window, kernel).unwrap();
     let first = chunk(&[(5, vec![Some(1.0)]), (12, vec![Some(2.0)])], 1, 0, false);
     let second = chunk(&[(18, vec![Some(4.0)]), (25, vec![Some(8.0)])], 1, 0, true);
     assert!(evaluator.push(&first).unwrap().is_none());
@@ -498,13 +494,17 @@ fn empty_input_and_series_without_results() {
 
 #[test]
 fn rejects_invalid_parameters() {
-    let kernel = || -> WindowKernel { Box::new(probe) };
     let zero_range = RangeWindow {
         range_ms: 0,
         offset_ms: 0,
     };
     assert!(matches!(
-        RangeEvaluator::try_new(&raw_schema(1), grid(0, 10, 4), zero_range, kernel()),
+        RangeEvaluator::try_new(
+            &raw_schema(1),
+            grid(0, 10, 4),
+            zero_range,
+            RangeFunction::Rate
+        ),
         Err(DataFusionError::Plan(_))
     ));
 
@@ -515,7 +515,7 @@ fn rejects_invalid_parameters() {
         offset_ms: 0,
     };
     assert!(matches!(
-        RangeEvaluator::try_new(&evaluated, grid(0, 10, 4), window, kernel()),
+        RangeEvaluator::try_new(&evaluated, grid(0, 10, 4), window, RangeFunction::Rate),
         Err(DataFusionError::Plan(_))
     ));
 }