diff --git a/src/promql/src/functions/aggr_over_time.rs b/src/promql/src/functions/aggr_over_time.rs
index 8f3b0703009..4a720fb19c6 100644
--- a/src/promql/src/functions/aggr_over_time.rs
+++ b/src/promql/src/functions/aggr_over_time.rs
@@ -102,9 +102,9 @@ fn evaluate_presence(
         .unwrap();
     // Windows overlap heavily, so locate every sample once and let each window binary-search
     // its own slice of the index instead of rescanning the slots it shares with its neighbours.
-    // Built only for a nullable batch, which is also the only branch that consults it.
     let has_nulls = values.null_count() != 0;
-    let valid_positions: Vec<usize> = if has_nulls {
+    let valid_positions: Vec<usize> = if has_nulls && matches!(operation, PresenceEvaluator::Count)
+    {
         (0..values.len())
             .filter(|&index| values.is_valid(index))
             .collect()
@@ -112,25 +112,25 @@ fn evaluate_presence(
         Vec::new()
     };
     let evaluator: fn(&Float64Array, &[usize], usize, usize) -> Option<f64> = if has_nulls {
-        // `valid_positions` only ever names samples, so the payloads behind them are the ones
-        // the null-free branch would have read, and no null slot is ever loaded.
         match operation {
             PresenceEvaluator::Count => |_, samples, offset, length| {
                 let count = valid_slot_bounds(samples, offset, offset + length).len();
                 (count != 0).then_some(count as f64)
             },
-            PresenceEvaluator::Last => |values, samples, offset, length| {
-                let bounds = valid_slot_bounds(samples, offset, offset + length);
-                (!bounds.is_empty()).then(|| values.value(samples[bounds.end - 1]))
+            PresenceEvaluator::Last => |values, _, offset, length| {
+                (offset..offset + length)
+                    .rev()
+                    .find(|&index| values.is_valid(index))
+                    .map(|index| values.value(index))
             },
-            PresenceEvaluator::Absent => |_, samples, offset, length| {
-                valid_slot_bounds(samples, offset, offset + length)
-                    .is_empty()
+            PresenceEvaluator::Absent => |values, _, offset, length| {
+                (!(offset..offset + length).any(|index| values.is_valid(index))).then_some(1.0)
+            },
+            PresenceEvaluator::Present => |values, _, offset, length| {
+                (offset..offset + length)
+                    .any(|index| values.is_valid(index))
                     .then_some(1.0)
             },
-            PresenceEvaluator::Present => |_, samples, offset, length| {
-                (!valid_slot_bounds(samples, offset, offset + length).is_empty()).then_some(1.0)
-            },
         }
     } else {
         match operation {
@@ -1313,42 +1313,6 @@ mod test {
         }
     }
 
-    /// `// oracle: pre-index implementation`
-    ///
-    /// The per-window scan the nullable presence evaluators performed before the valid slots
-    /// were indexed.
-    fn oracle_presence_output(
-        values: &Float64Array,
-        ranges: &[(u32, u32)],
-        operation: PresenceEvaluator,
-    ) -> Vec<Option<f64>> {
-        ranges
-            .iter()
-            .map(|&(offset, length)| {
-                let offset = offset as usize;
-                let end = offset + length as usize;
-                match operation {
-                    PresenceEvaluator::Count => {
-                        let count = (offset..end)
-                            .filter(|&index| values.is_valid(index))
-                            .count();
-                        (count != 0).then_some(count as f64)
-                    }
-                    PresenceEvaluator::Last => (offset..end)
-                        .rev()
-                        .find(|&index| values.is_valid(index))
-                        .map(|index| values.value(index)),
-                    PresenceEvaluator::Absent => {
-                        (!(offset..end).any(|index| values.is_valid(index))).then_some(1.0)
-                    }
-                    PresenceEvaluator::Present => (offset..end)
-                        .any(|index| values.is_valid(index))
-                        .then_some(1.0),
-                }
-            })
-            .collect()
-    }
-
     /// Payload kept under the null slots of the layouts below, so an evaluator that read the
     /// padding instead of the samples would report a different value for `last_over_time`.
     const NULL_PAYLOAD: f64 = -1234.5;
@@ -1395,13 +1359,9 @@ mod test {
     /// backing array.
     fn presence_window_shapes() -> Vec<(&'static str, Vec<(u32, u32)>)> {
         vec![
-            // Heavily overlapping: neighbours share eleven of their twelve slots.
             ("overlapping", (0..=12).map(|i| (i, 12)).collect()),
-            // Small windows, each overlapping its neighbour by two slots.
             ("small", (0..21).map(|i| (i, 3)).collect()),
-            // Disjoint: no slot is shared with another window.
             ("disjoint", (0..6).map(|i| (i * 4, 4)).collect()),
-            // Empty windows hold no sample at any offset.
             ("empty", vec![(0, 0), (12, 0), (23, 0)]),
             // Boundaries on both sides of a null run, and windows sitting inside one.
             (
@@ -1447,56 +1407,36 @@ mod test {
             .collect()
     }
 
-    /// Asserts two `Option<f64>` sequences agree bit for bit, naming the layout in failures.
-    fn assert_presence_matches_oracle(label: &str, actual: &[Option<f64>], oracle: &[Option<f64>]) {
-        assert_eq!(actual.len(), oracle.len(), "{label}");
-        for (index, (actual, oracle)) in actual.iter().zip(oracle).enumerate() {
-            let equal = match (actual, oracle) {
-                (None, None) => true,
-                (Some(actual), Some(oracle)) => actual.to_bits() == oracle.to_bits(),
-                _ => false,
-            };
-            assert!(
-                equal,
-                "{label}: window {index}: indexed {actual:?} != oracle {oracle:?}"
-            );
-        }
-    }
-
     #[test]
     fn presence_range_udfs_match_per_window_oracle_on_nullable_layouts() {
-        // Five backing layouts times five window shapes, each checked against the pre-index
-        // per-window scan for all four presence functions.
+        // The scalar kernels independently validate the specialized UDFs.
         let len = 24;
-        // Constructors rather than `ScalarUDF` values, so the table is `Copy` and the loops
-        // below can walk it once per layout.
-        type PresenceCase = (&'static str, fn() -> ScalarUDF, PresenceEvaluator);
+        type PresenceCase = (
+            fn() -> ScalarUDF,
+            fn(&TimestampMillisecondArray, &Float64Array) -> Option<f64>,
+        );
         let cases: [PresenceCase; 4] = [
-            ("count", CountOverTime::scalar_udf, PresenceEvaluator::Count),
-            ("last", LastOverTime::scalar_udf, PresenceEvaluator::Last),
-            (
-                "absent",
-                AbsentOverTime::scalar_udf,
-                PresenceEvaluator::Absent,
-            ),
-            (
-                "present",
-                PresentOverTime::scalar_udf,
-                PresenceEvaluator::Present,
-            ),
+            (CountOverTime::scalar_udf, count_over_time),
+            (LastOverTime::scalar_udf, last_over_time),
+            (AbsentOverTime::scalar_udf, absent_over_time),
+            (PresentOverTime::scalar_udf, present_over_time),
         ];
 
-        for (layout, nulls) in presence_layouts(len) {
-            for (shape, ranges) in presence_window_shapes() {
+        for (_, nulls) in presence_layouts(len) {
+            for (_, ranges) in presence_window_shapes() {
                 let values = nullable_backing_values(&nulls);
-                for (name, build_udf, operation) in cases {
-                    let oracle = oracle_presence_output(&values, &ranges, operation);
+                for (build_udf, kernel) in cases {
+                    let expected = ranges
+                        .iter()
+                        .map(|&(offset, length)| {
+                            let values = values.slice(offset as usize, length as usize);
+                            let values = values.as_any().downcast_ref::<Float64Array>().unwrap();
+                            let timestamps = TimestampMillisecondArray::new_null(values.len());
+                            kernel(&timestamps, values)
+                        })
+                        .collect::<Vec<_>>();
                     let actual = run_presence_udf(build_udf(), &values, &ranges);
-                    assert_presence_matches_oracle(
-                        &format!("{name}, {layout}, {shape}"),
-                        &actual,
-                        &oracle,
-                    );
+                    assert_option_bits(&actual, &expected);
                 }
             }
         }
@@ -1521,7 +1461,6 @@ mod test {
         let cases = [
             (
                 CountOverTime::scalar_udf(),
-                PresenceEvaluator::Count,
                 vec![
                     Some(3.0),
                     Some(2.0),
@@ -1535,7 +1474,6 @@ mod test {
             ),
             (
                 LastOverTime::scalar_udf(),
-                PresenceEvaluator::Last,
                 vec![
                     Some(9.0),
                     Some(3.0),
@@ -1549,7 +1487,6 @@ mod test {
             ),
             (
                 AbsentOverTime::scalar_udf(),
-                PresenceEvaluator::Absent,
                 vec![
                     None,
                     None,
@@ -1563,7 +1500,6 @@ mod test {
             ),
             (
                 PresentOverTime::scalar_udf(),
-                PresenceEvaluator::Present,
                 vec![
                     Some(1.0),
                     Some(1.0),
@@ -1577,11 +1513,9 @@ mod test {
             ),
         ];
 
-        for (udf, operation, expected) in cases {
+        for (udf, expected) in cases {
             let actual = run_presence_udf(udf, &values, &ranges);
             assert_option_bits(&actual, &expected);
-            let oracle = oracle_presence_output(&values, &ranges, operation);
-            assert_presence_matches_oracle("null runs", &actual, &oracle);
         }
     }
 
diff --git a/src/promql/src/functions/extrapolate_rate.rs b/src/promql/src/functions/extrapolate_rate.rs
index 4a375b9eb9e..78d9704db27 100644
--- a/src/promql/src/functions/extrapolate_rate.rs
+++ b/src/promql/src/functions/extrapolate_rate.rs
@@ -192,7 +192,6 @@ impl<const IS_COUNTER: bool, const IS_RATE: bool> ExtrapolatedRate<IS_COUNTER, I
         let has_nulls = value_array.null_count() > 0;
         // Windows overlap heavily, so locate every sample once and let each window find its own
         // slice of the index, instead of rescanning the slots it shares with its neighbours.
-        // Built only for a nullable batch, which is also the only branch that consults it.
         let valid_positions: Vec<usize> = if has_nulls {
             (0..value_array.len())
                 .filter(|&index| value_array.is_valid(index))
@@ -770,11 +769,7 @@ mod test {
         assert_eq!(output, vec![Some(7.5)]);
     }
 
-    /// `// oracle: pre-index implementation`
-    ///
-    /// The window computation as it stood before the valid-slot index existed: every window
-    /// rescanned its own slots for the samples and for the counter resets. Same math as
-    /// [`ExtrapolatedRate::calc`], reached the long way.
+    /// This independent scan implementation validates the indexed nullable path.
     fn oracle_rate_output<const IS_COUNTER: bool, const IS_RATE: bool>(
         timestamps: &[i64],
         values: &Float64Array,
@@ -910,45 +905,6 @@ mod test {
         }
     }
 
-    /// [`assert_rate_matches_oracle`] for `0` = rate, `1` = increase and `2` = delta.
-    fn assert_rate_variant_matches_oracle(
-        variant: usize,
-        label: &str,
-        timestamps: Vec<i64>,
-        values: Arc<Float64Array>,
-        ranges: Vec<(u32, u32)>,
-        eval_timestamps: Vec<i64>,
-        range_length: i64,
-    ) {
-        match variant {
-            0 => assert_rate_matches_oracle::<true, true>(
-                label,
-                timestamps,
-                values,
-                ranges,
-                eval_timestamps,
-                range_length,
-            ),
-            1 => assert_rate_matches_oracle::<true, false>(
-                label,
-                timestamps,
-                values,
-                ranges,
-                eval_timestamps,
-                range_length,
-            ),
-            2 => assert_rate_matches_oracle::<false, false>(
-                label,
-                timestamps,
-                values,
-                ranges,
-                eval_timestamps,
-                range_length,
-            ),
-            _ => unreachable!("only rate, increase and delta exist"),
-        }
-    }
-
     /// Payload kept under the null slots of the layouts below, so a window that read the
     /// padding instead of the samples would report a different value.
     const NULL_PAYLOAD: f64 = -1234.5;
@@ -1017,8 +973,8 @@ mod test {
 
     #[test]
     fn nullable_rate_windows_match_per_window_oracle() {
-        // Four layouts times three window shapes times the three functions, every window of
-        // every batch checked against the pre-index implementation.
+        // Four layouts times three window shapes times the three functions, every window
+        // checked against the independent scan implementation.
         let len = 24;
         let timestamps: Vec<i64> = (0..len)
             .map(|index| index as i64 * 30_000 + (index % 5) as i64 * 1_000)
@@ -1028,9 +984,20 @@ mod test {
         for (input, nulls) in matrix_inputs(len) {
             for (shape, ranges) in matrix_window_shapes(len) {
                 let eval_timestamps = eval_timestamps_after(&timestamps, &ranges, 5_000);
-                for (variant, name) in ["rate", "increase", "delta"].into_iter().enumerate() {
-                    assert_rate_variant_matches_oracle(
-                        variant,
+                type CheckFn =
+                    fn(&str, Vec<i64>, Arc<Float64Array>, Vec<(u32, u32)>, Vec<i64>, i64);
+                for (name, check) in [
+                    ("rate", assert_rate_matches_oracle::<true, true> as CheckFn),
+                    (
+                        "increase",
+                        assert_rate_matches_oracle::<true, false> as CheckFn,
+                    ),
+                    (
+                        "delta",
+                        assert_rate_matches_oracle::<false, false> as CheckFn,
+                    ),
+                ] {
+                    check(
                         &format!("{name}, {input}, {shape}"),
                         timestamps.clone(),
                         values_with_nulls(nulls.clone(), NULL_PAYLOAD),
@@ -1104,9 +1071,13 @@ mod test {
                 .collect();
 
             let label = format!("seed {seed:#x}");
-            for variant in 0..3 {
-                assert_rate_variant_matches_oracle(
-                    variant,
+            type CheckFn = fn(&str, Vec<i64>, Arc<Float64Array>, Vec<(u32, u32)>, Vec<i64>, i64);
+            for check in [
+                assert_rate_matches_oracle::<true, true> as CheckFn,
+                assert_rate_matches_oracle::<true, false> as CheckFn,
+                assert_rate_matches_oracle::<false, false> as CheckFn,
+            ] {
+                check(
                     &label,
                     timestamps.clone(),
                     values.clone(),
@@ -1149,33 +1120,22 @@ mod test {
         // increase of 4.0 over two intervals is extrapolated by half an interval to 5.0. Reading
         // the null padding as a sample would charge a reset against it and change both values.
         let output = nullable_rate_runner::<true, false>(
-            timestamps.clone(),
-            values(),
-            ranges.clone(),
-            eval_timestamps.clone(),
-            60_000,
-        );
-        assert_eq!(
-            output,
-            vec![None, None, Some(1.5), Some(5.0), Some(5.0), None, None]
-        );
-
-        assert_rate_variant_matches_oracle(
-            1,
-            "increase, windows cutting through nulls",
             timestamps,
             values(),
             ranges,
             eval_timestamps,
             60_000,
         );
+        assert_eq!(
+            output,
+            vec![None, None, Some(1.5), Some(5.0), Some(5.0), None, None]
+        );
     }
 
     #[test]
     fn nullable_rate_keeps_special_values_out_of_the_null_branch() {
         // NaN, infinities and stale markers are valid samples, not missing ones: the windows
-        // keep counting them, and only the null slots stay out of the sample set. The result
-        // is NaN because the samples are, and the oracle agrees.
+        // keep counting them, and only the null slots stay out of the sample set.
         let timestamps: Vec<i64> = (0..6).map(|index| index as i64 * 1000).collect();
         let values = || {
             values_with_nulls(
@@ -1194,10 +1154,10 @@ mod test {
         let eval_timestamps = eval_timestamps_after(&timestamps, &ranges, 0);
 
         let output = nullable_rate_runner::<false, false>(
-            timestamps.clone(),
+            timestamps,
             values(),
-            ranges.clone(),
-            eval_timestamps.clone(),
+            ranges,
+            eval_timestamps,
             60_000,
         );
         // The first two windows hold at least two valid samples, so they report a value, NaN
@@ -1207,16 +1167,6 @@ mod test {
         assert!(output[1].is_some_and(|value| value.is_nan()));
         assert_eq!(output[2], None);
         assert_eq!(output[3], None);
-
-        assert_rate_variant_matches_oracle(
-            2,
-            "delta, special values",
-            timestamps,
-            values(),
-            ranges,
-            eval_timestamps,
-            60_000,
-        );
     }
 
     /// Line-by-line port of Prometheus `extrapolatedRate` (promql/functions.go), float path