@@ -51,8 +51,6 @@ pub struct StoreQueryParams {
5151 pub start_timestamp : u64 ,
5252 /// Milliseconds since epoch.
5353 pub end_timestamp : u64 ,
54- /// true for sliding windows (exact match), false for tumbling (range)
55- pub is_exact_query : bool ,
5654}
5755
5856/// Complete plan for querying store (values + optional separate keys)
@@ -79,6 +77,9 @@ pub struct QueryExecutionContext {
7977 pub metadata : QueryMetadata ,
8078 pub store_plan : StoreQueryPlan ,
8179 pub agg_info : AggregationIdInfo ,
80+ /// The value aggregation's WindowType -- Sliding fetches/merges a single
81+ /// already-complete window; Tumbling sums the disjoint buckets in range.
82+ pub value_window_type : WindowType ,
8283 /// Whether to merge multiple precomputes (true for temporal queries)
8384 pub do_merge : bool ,
8485 #[ allow( dead_code) ]
@@ -416,24 +417,33 @@ impl SimpleEngine {
416417 }
417418 } ;
418419
420+ // Keys always fetch via the window-grid walk (execute_store_query),
421+ // never a single exact-window lookup -- this is an explicit,
422+ // permanent choice, not a WindowType derivation: a keys query
423+ // conceptually always needs to see the key's own bucket(s), not "the
424+ // one window ending now."
419425 Ok ( StoreQueryParams {
420426 metric : metric. to_string ( ) ,
421427 aggregation_id : agg_info. aggregation_id_for_key ,
422428 start_timestamp,
423429 end_timestamp,
424- is_exact_query : false , // Keys always use range queries
425430 } )
426431 }
427432
428433 /// Creates a plan for querying the store based on aggregation configuration.
429434 /// Also derives `do_merge`: true when the requested time range spans more
430435 /// than one stored window, i.e. `range_ms > window_size_ms`.
436+ ///
437+ /// Returns the value aggregation's `WindowType` alongside the plan --
438+ /// callers need it again later (e.g. to pick merge semantics) and it's
439+ /// cheaper to hand back what was already looked up here than to
440+ /// re-fetch the aggregation config.
431441 fn create_store_query_plan (
432442 & self ,
433443 metric : & str ,
434444 timestamps : & QueryTimestamps ,
435445 agg_info : & AggregationIdInfo ,
436- ) -> Result < ( StoreQueryPlan , bool ) , String > {
446+ ) -> Result < ( StoreQueryPlan , bool , WindowType ) , String > {
437447 let sc = self . streaming_config . read ( ) . unwrap ( ) . clone ( ) ;
438448 // Get aggregation config for value to determine window type
439449 let aggregation_config_for_value = sc
@@ -446,13 +456,15 @@ impl SimpleEngine {
446456 } ) ?;
447457
448458 let window_type = aggregation_config_for_value. window_type ;
449- let is_exact_query = window_type == WindowType :: Sliding ;
450459 let range_ms = timestamps. end_timestamp - timestamps. start_timestamp ;
451460 let do_merge = range_ms > aggregation_config_for_value. window_size_ms ;
452461
453- // Determine start/end for values query based on window type
454- let ( values_start, values_end) = if is_exact_query {
455- // Sliding window: exact window match
462+ // Determine start/end for values query based on window type. For
463+ // Sliding, narrow to exactly the one window ending "now" --
464+ // execute_store_query's window-grid walk degenerates to a single
465+ // exact lookup when given a range exactly one window wide, so this
466+ // narrowing (not a separate flag) is what makes it an "exact" fetch.
467+ let ( values_start, values_end) = if window_type == WindowType :: Sliding {
456468 let exact_start =
457469 timestamps. end_timestamp - aggregation_config_for_value. window_size_ms ;
458470 ( exact_start, timestamps. end_timestamp )
@@ -466,7 +478,6 @@ impl SimpleEngine {
466478 aggregation_id : agg_info. aggregation_id_for_value ,
467479 start_timestamp : values_start,
468480 end_timestamp : values_end,
469- is_exact_query,
470481 } ;
471482
472483 // Determine if we need a separate keys query
@@ -482,6 +493,7 @@ impl SimpleEngine {
482493 keys_query,
483494 } ,
484495 do_merge,
496+ window_type,
485497 ) )
486498 }
487499
@@ -504,13 +516,14 @@ impl SimpleEngine {
504516 }
505517 }
506518
507- /// Non-exact store query: walks the aggregation's window grid
508- /// (`bucket_step_ms` apart, each window `window_size_ms` wide, per
509- /// `WindowManager::window_start_for`) and looks up every grid position
510- /// in `[start_timestamp, end_timestamp)` with an exact match, merging
511- /// the sparse per-window results. Used for range queries, key queries,
512- /// and instant queries over tumbling windows — everywhere
513- /// `is_exact_query` is false.
519+ /// Walks the aggregation's window grid (`bucket_step_ms` apart, each
520+ /// window `window_size_ms` wide, per `WindowManager::window_start_for`)
521+ /// and looks up every grid position in `[start_timestamp, end_timestamp)`
522+ /// with an exact match, merging the sparse per-window results. A range
523+ /// exactly one window wide degenerates to a single exact lookup -- an
524+ /// instant Sliding-window fetch gets "the one window ending now" this
525+ /// way, by being narrowed to one window's width before calling
526+ /// (`create_store_query_plan`), not via a separate exact/scan flag.
514527 fn scan_windows_via_exact (
515528 & self ,
516529 params : & StoreQueryParams ,
@@ -588,65 +601,20 @@ impl SimpleEngine {
588601 params : & StoreQueryParams ,
589602 ) -> Result < TimestampedBucketsMap , String > {
590603 debug ! (
591- "Querying store: metric={}, agg_id={}, range=[{}, {}], exact={}" ,
592- params. metric,
593- params. aggregation_id,
594- params. start_timestamp,
595- params. end_timestamp,
596- params. is_exact_query
604+ "Querying store: metric={}, agg_id={}, range=[{}, {}]" ,
605+ params. metric, params. aggregation_id, params. start_timestamp, params. end_timestamp,
597606 ) ;
598607
599608 let store_query_start_time = Instant :: now ( ) ;
600-
601- let result = if params. is_exact_query {
602- debug ! (
603- "Sliding window query: Looking for exact window [{}, {}]" ,
604- params. start_timestamp, params. end_timestamp
605- ) ;
606- let res = self
607- . store
608- . query_precomputed_output_exact (
609- & params. metric ,
610- params. aggregation_id ,
611- params. start_timestamp ,
612- params. end_timestamp ,
613- )
614- . map_err ( |e| {
615- format ! (
616- "Error querying store for metric {}, agg {}, range [{}, {}]: {}" ,
617- params. metric,
618- params. aggregation_id,
619- params. start_timestamp,
620- params. end_timestamp,
621- e
622- )
623- } ) ;
624- if let Ok ( ref outputs) = res {
625- let store_query_duration = store_query_start_time. elapsed ( ) ;
626- debug ! (
627- "Sliding window exact query took: {:.2}ms, found {} unique keys" ,
628- store_query_duration. as_secs_f64( ) * 1000.0 ,
629- outputs. len( )
630- ) ;
631- }
632- res
633- } else {
609+ let result = self . scan_windows_via_exact ( params) ;
610+ if let Ok ( ref outputs) = result {
611+ let store_query_duration = store_query_start_time. elapsed ( ) ;
634612 debug ! (
635- "Window-grid query: range [{}, {}]" ,
636- params. start_timestamp, params. end_timestamp
613+ "Window-grid query took: {:.2}ms, found {} unique keys" ,
614+ store_query_duration. as_secs_f64( ) * 1000.0 ,
615+ outputs. len( )
637616 ) ;
638- let res = self . scan_windows_via_exact ( params) ;
639- if let Ok ( ref outputs) = res {
640- let store_query_duration = store_query_start_time. elapsed ( ) ;
641- debug ! (
642- "Window-grid query took: {:.2}ms, found {} unique keys" ,
643- store_query_duration. as_secs_f64( ) * 1000.0 ,
644- outputs. len( )
645- ) ;
646- }
647- res
648- } ;
649-
617+ }
650618 result
651619 }
652620
@@ -656,6 +624,7 @@ impl SimpleEngine {
656624 plan : & StoreQueryPlan ,
657625 do_merge : bool ,
658626 agg_info : & AggregationIdInfo ,
627+ value_window_type : WindowType ,
659628 ) -> Result < ( MergedOutputsMap , Option < MergedOutputsMap > ) , String > {
660629 // Query and merge values
661630 let values_map = self . execute_store_query ( & plan. values_query ) . map_err ( |e| {
@@ -673,13 +642,8 @@ impl SimpleEngine {
673642 debug ! ( "Store query returned {} unique keys" , values_map. len( ) ) ;
674643
675644 let merge_start_time = Instant :: now ( ) ;
676- let window_type = if plan. values_query . is_exact_query {
677- WindowType :: Sliding
678- } else {
679- WindowType :: Tumbling
680- } ;
681645
682- let merged_values = if plan . values_query . is_exact_query {
646+ let merged_values = if value_window_type == WindowType :: Sliding {
683647 // Sliding window: expected exactly 1 precompute per key today
684648 // (ponytail: hardcoded, #554 will make >1 legitimate — don't
685649 // block on it). The store can legitimately return more than
@@ -714,7 +678,7 @@ impl SimpleEngine {
714678 } ;
715679
716680 let merge_duration = merge_start_time. elapsed ( ) ;
717- let did_merge = window_type == WindowType :: Sliding
681+ let did_merge = value_window_type == WindowType :: Sliding
718682 || do_merge
719683 || agg_info. aggregation_type_for_value == AggregationType :: DeltaSetAggregator ;
720684 debug ! (
@@ -895,6 +859,7 @@ impl SimpleEngine {
895859 & context. store_plan ,
896860 context. do_merge ,
897861 & context. agg_info ,
862+ context. value_window_type ,
898863 ) ?;
899864
900865 // Step 2: Collect results
0 commit comments