test(query-engine): cover SQL dual populations

Author: milindsrivastava1997

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -777,6 +777,9 @@ impl SimpleEngine {
             window_size_ms,
             slide_interval_ms,
         };
+        // Range-step covers overlap heavily. Keep one sorted set so each
+        // exact stored window is fetched once, in deterministic order, even
+        // when multiple output timestamps require it.
         let mut windows = BTreeSet::new();
         for &output_timestamp in output_timestamps {
             let cover = plan_exact_cover(output_timestamp, lookback_ms, spec).map_err(|error| {

asap-query-engine/src/tests/native_range_query_tests.rs

@@ -257,11 +257,8 @@ mod tests {
     /// but the KEY aggregation is a Sliding window with
     /// `key_slide_interval_ms < key_window_size_ms` (#600). Real Sliding
     /// buckets are persisted on the slide_interval_ms grid, not the
-    /// window_size_ms grid (`precompute_engine/window_manager.rs`), so the
-    /// keys bucket span here is `key_slide_interval_ms`, not
-    /// `key_window_size_ms` -- unlike the value side, which stays Tumbling
-    /// (span == window) exactly as `create_range_engine_dual_input_with_windows`
-    /// already does.
+    /// window_size_ms grid (`precompute_engine/window_manager.rs`), and the
+    /// value side uses the same Sliding grid for these tests.
     #[allow(clippy::too_many_arguments)]
     fn create_range_engine_dual_input_sliding_keys(
         metric: &str,
@@ -411,8 +408,8 @@ mod tests {
         //   t=5000, whose keys lookback window is
         //   [5000 - key_window_size_ms, 5000) = [3000,5000) -- lining up
         //   exactly with the inserted bucket.
-        // - value_data is also placed at timestamp=5000 (Tumbling, 1000ms
-        //   wide -> bucket [4000,5000)) purely so the CountMinSketch value
+        // - value_data is also placed at timestamp=5000 (Sliding, 2000ms
+        //   wide -> bucket [3000,5000)) purely so the CountMinSketch value
         //   side resolves at the same t=5000 step; it's unrelated to #600.
         let mut keys_add = SetAggregatorAccumulator::new();
         keys_add.add_key(KeyByLabelValues {

asap-query-engine/src/tests/query_equivalence_tests.rs

@@ -137,6 +137,88 @@ mod tests {
         assert_eq!(vector.values[0].value, 18.0);
     }
 
+    #[test]
+    fn sql_executes_wider_sliding_dual_population_query() {
+        use crate::data_model::{
+            AggregationConfig, AggregationType, CleanupPolicy, KeyByLabelValues, PrecomputedOutput,
+        };
+        use crate::engines::query_result::QueryResult;
+        use crate::precompute_operators::{CountMinSketchAccumulator, SetAggregatorAccumulator};
+        use crate::stores::simple_map_store::SimpleMapStore;
+
+        let sql_query = "SELECT COUNT(value) FROM events WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY host";
+        let (_, mut sql_config, mut streaming_config) = TestConfigBuilder::new("events")
+            .with_grouping_labels(vec!["host"])
+            .add_temporal_query(
+                "count_over_time(events[10s])",
+                sql_query,
+                1,
+                5_000,
+                WindowType::Sliding,
+            )
+            .build_both();
+
+        let value_config = streaming_config
+            .get_aggregation_config(1)
+            .expect("value config")
+            .clone();
+        let mut value_config = AggregationConfig {
+            aggregation_type: AggregationType::CountMinSketch,
+            ..value_config
+        };
+        value_config.slide_interval_ms = 1_000;
+        let mut key_config = value_config.clone();
+        key_config.aggregation_id = 2;
+        key_config.aggregation_type = AggregationType::SetAggregator;
+        streaming_config = Arc::new(crate::data_model::StreamingConfig {
+            aggregation_configs: HashMap::from([(1, value_config), (2, key_config)]),
+        });
+        sql_config.query_configs[1] = sql_config.query_configs[1]
+            .clone()
+            .add_aggregation(crate::data_model::AggregationReference::new(2, None));
+
+        let store = Arc::new(SimpleMapStore::new(
+            streaming_config.clone(),
+            CleanupPolicy::NoCleanup,
+        ));
+        for (start, end, count) in [(0, 5_000, 2.0), (5_000, 10_000, 3.0)] {
+            let mut value = CountMinSketchAccumulator::new(4, 64);
+            value.inner.update("host-a", count);
+            let mut keys = SetAggregatorAccumulator::new();
+            keys.add_key(KeyByLabelValues {
+                labels: vec!["host-a".to_string()],
+            });
+            store
+                .insert_precomputed_output(
+                    PrecomputedOutput::new(start, end, None, 1),
+                    Box::new(value),
+                )
+                .unwrap();
+            store
+                .insert_precomputed_output(
+                    PrecomputedOutput::new(start, end, None, 2),
+                    Box::new(keys),
+                )
+                .unwrap();
+        }
+
+        let engine = SimpleEngine::new(
+            store,
+            sql_config,
+            streaming_config,
+            1_000,
+            QueryLanguage::sql,
+        );
+        let (_, result) = engine
+            .handle_query_sql(sql_query.to_string(), 10.0)
+            .expect("SQL dual-population query should execute");
+        let QueryResult::Vector(vector) = result else {
+            panic!("expected an instant SQL vector");
+        };
+        assert_eq!(vector.values.len(), 1);
+        assert_eq!(vector.values[0].value, 5.0);
+    }
+
     #[test]
     fn test_temporal_sum_equivalence() {
         let scrape_interval_ms = 1000;