Fix historical sliding window semantics

benches/historical_sliding.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 use support::{populate_storage, recent_base_timestamp, unique_config, GRAPH_URI};
 
 // Window config: OFFSET=10_000ms, RANGE=2_000ms, SLIDE=1_000ms
-// SlidingWindowIterator scans [now-10000, now] with 8 overlapping windows.
+// SlidingWindowIterator covers [now-10000, now) with 9 overlapping windows.
 // Data is written at [now-8000, now-2000] — solidly within the scan range.
 const OFFSET_MS: u64 = 10_000;
 const RANGE_MS: u64 = 2_000;

docs/JANUSQL.md

@@ -54,11 +54,13 @@ FROM NAMED WINDOW ex:previousHour ON LOG ex:log [OFFSET 86400000 RANGE 3600000 S
 At evaluation time `T`, the window is:
 
 ```text
-[T - OFFSET - RANGE, T - OFFSET]
+[T - OFFSET, T - OFFSET + RANGE)
 ```
 
-The range must not exceed the offset; Janus rejects a sliding historical
-window that would extend beyond its evaluation time.
+`OFFSET` determines how far before `T` the historical window starts, and
+`RANGE` determines its duration. The range must not exceed the offset; Janus
+rejects a sliding historical window that would extend beyond its evaluation
+time.
 
 ### Sliding live window
 

docs/WINDOW_TYPES_EXPLAINED.md

@@ -17,8 +17,10 @@ Evaluates once over a persisted event-log interval. `END` must be later than
 FROM NAMED WINDOW ex:previousHour ON LOG ex:log [OFFSET 86400000 RANGE 3600000 STEP 30000]
 ```
 
-At time `T`, Janus evaluates `[T - OFFSET - RANGE, T - OFFSET]`. The range
-cannot exceed the offset.
+At time `T`, Janus evaluates the half-open interval
+`[T - OFFSET, T - OFFSET + RANGE)`. `OFFSET` determines the historical start
+relative to `T`, and `RANGE` determines the interval duration. The range cannot
+exceed the offset, which keeps the interval end at or before `T`.
 
 ## Live sliding
 

src/api/janus_api/baseline.rs

@@ -253,11 +253,19 @@ pub(crate) fn load_or_compute_baseline_snapshot(
                         definition.name, source_window.window_name
                     ))
                 })?;
-            executor.execute_window_bounds(
-                window_start,
-                window_end,
-                &generated_query.sparql_query,
-            )?
+            if source_window.window_type == WindowType::HistoricalSliding {
+                executor.execute_window_bounds_half_open(
+                    window_start,
+                    window_end,
+                    &generated_query.sparql_query,
+                )?
+            } else {
+                executor.execute_window_bounds(
+                    window_start,
+                    window_end,
+                    &generated_query.sparql_query,
+                )?
+            }
         }
         _ => executor.execute_materialized_historical_subquery(
             &source_windows,

src/api/janus_api/tests.rs

@@ -784,14 +784,14 @@ HAVING(AVG(?value) > ?yesterdayAvgValue)
     let latest_rows = Arc::new(RwLock::new(HashMap::new()));
     assert_eq!(
         storage
-            .query_rdf(86_400_001, 86_460_001)
+            .query_rdf_half_open(86_400_001, 86_460_001)
             .expect("first historical range should query")
             .len(),
         2
     );
     assert_eq!(
         storage
-            .query_rdf(86_460_001, 86_520_001)
+            .query_rdf_half_open(86_460_001, 86_520_001)
             .expect("second historical range should query")
             .len(),
         2

src/bin/hybrid_scaling_combined.rs

@@ -14,6 +14,7 @@ use janus::paper_bench::query_defined_baseline::rdf::{
 };
 use janus::parsing::janusql_parser::{
     BaselineDefinition, BaselineGraphTemplate, JanusQLParser, ParsedJanusQuery, WindowDefinition,
+    WindowType,
 };
 use janus::storage::segmented_storage::StreamingSegmentedStorage;
 use janus::storage::util::StreamingConfig;
@@ -884,7 +885,11 @@ fn execute_lowered_historical_subquery(
         let (start, end) = window.resolve_historical_bounds(evaluation_time).ok_or_else(|| {
             format!("failed to resolve historical bounds for window '{}'", window.window_name)
         })?;
-        historical_events.extend(storage.query_rdf(start, end)?);
+        historical_events.extend(if window.window_type == WindowType::HistoricalSliding {
+            storage.query_rdf_half_open(start, end)?
+        } else {
+            storage.query_rdf(start, end)?
+        });
     }
 
     Ok(build_historical_baseline_bindings_from_events(&historical_events))

src/execution/historical_executor.rs

@@ -19,7 +19,6 @@ use crate::parsing::janusql_parser::WindowDefinition;
 use crate::querying::oxigraph_adapter::OxigraphAdapter;
 use crate::storage::segmented_storage::StreamingSegmentedStorage;
 use crate::stream::operators::historical_fixed_window::HistoricalFixedWindowOperator;
-use crate::stream::operators::historical_sliding_window::HistoricalSlidingWindowOperator;
 use oxigraph::model::{GraphName, NamedNode, Quad};
 use rsp_rs::QuadContainer;
 use std::collections::{HashMap, HashSet};
@@ -103,6 +102,24 @@ impl HistoricalExecutor {
         self.execute_sparql_on_events(&events, sparql_query)
     }
 
+    /// Execute a query over an explicitly supplied half-open historical range.
+    ///
+    /// Fixed historical ranges retain the storage API's inclusive end behavior;
+    /// this variant is for resolved sliding-window bounds.
+    pub fn execute_window_bounds_half_open(
+        &self,
+        start: u64,
+        end: u64,
+        sparql_query: &str,
+    ) -> Result<Vec<HashMap<String, String>>, JanusApiError> {
+        let events = self
+            .storage
+            .query_half_open(start, end)
+            .map_err(|e| JanusApiError::StorageError(format!("Failed to query storage: {}", e)))?;
+
+        self.execute_sparql_on_events(&events, sparql_query)
+    }
+
     /// Execute one historical materialized result over one or more historical windows by
     /// loading each window into a synthetic named graph keyed by the JanusQL window name.
     pub fn execute_materialized_historical_subquery(
@@ -122,9 +139,21 @@ impl HistoricalExecutor {
                         window.window_name
                     ))
                 })?;
-            let events = self.storage.query(start, end).map_err(|e| {
-                JanusApiError::StorageError(format!("Failed to query storage: {}", e))
-            })?;
+            let events = match window.window_type {
+                crate::parsing::janusql_parser::WindowType::HistoricalSliding => {
+                    self.storage.query_half_open(start, end)
+                }
+                crate::parsing::janusql_parser::WindowType::HistoricalFixed => {
+                    self.storage.query(start, end)
+                }
+                crate::parsing::janusql_parser::WindowType::Live => {
+                    return Err(JanusApiError::ExecutionError(format!(
+                        "Window '{}' is not historical",
+                        window.window_name
+                    )))
+                }
+            }
+            .map_err(|e| JanusApiError::StorageError(format!("Failed to query storage: {}", e)))?;
             timestamps.extend(events.iter().map(|event| event.timestamp));
             let rdf_events = self.decode_events(&events)?;
             quads.extend(
@@ -166,23 +195,16 @@ impl HistoricalExecutor {
         window: &WindowDefinition,
         sparql_query: &'a str,
     ) -> impl Iterator<Item = Result<Vec<HashMap<String, String>>, JanusApiError>> + 'a {
-        let offset = window.offset.unwrap_or(0);
-        let width = window.width;
-        let slide = window.slide;
-
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
             .as_millis() as u64;
 
-        let start_time = now.saturating_sub(offset);
-
         SlidingWindowIterator {
             executor: self,
-            current_start: start_time,
-            evaluation_time: now,
-            width,
-            slide,
+            window: window.clone(),
+            current_evaluation_time: now,
+            latest_evaluation_time: now,
             sparql_query: sparql_query.to_string(),
         }
     }
@@ -397,52 +419,45 @@ impl HistoricalExecutor {
         &self,
         window: &WindowDefinition,
     ) -> Result<(u64, u64), JanusApiError> {
-        // For fixed windows: use explicit start/end
-        if let (Some(start), Some(end)) = (window.start, window.end) {
-            return Ok((start, end));
-        }
-
-        // For sliding windows: calculate from offset and width
-        if let Some(offset) = window.offset {
-            let now = std::time::SystemTime::now()
+        let evaluation_time = if window.offset.is_some() {
+            std::time::SystemTime::now()
                 .duration_since(std::time::UNIX_EPOCH)
                 .map_err(|e| JanusApiError::ExecutionError(format!("System time error: {}", e)))?
-                .as_millis() as u64;
-
-            let start = now.saturating_sub(offset);
-            let end = start + window.width;
-            return Ok((start, end));
-        }
+                .as_millis() as u64
+        } else {
+            window.end.unwrap_or_default()
+        };
 
-        Err(JanusApiError::ExecutionError(
-            "Window definition must have either (start, end) or (offset, width)".to_string(),
-        ))
+        window.resolve_historical_bounds(evaluation_time).ok_or_else(|| {
+            JanusApiError::ExecutionError(
+                "Window definition cannot resolve complete historical bounds".to_string(),
+            )
+        })
     }
 }
 
 /// Iterator for sliding windows that queries storage directly
 struct SlidingWindowIterator<'a> {
     executor: &'a HistoricalExecutor,
-    current_start: u64,
-    evaluation_time: u64,
-    width: u64,
-    slide: u64,
+    window: WindowDefinition,
+    current_evaluation_time: u64,
+    latest_evaluation_time: u64,
     sparql_query: String,
 }
 
 impl<'a> Iterator for SlidingWindowIterator<'a> {
     type Item = Result<Vec<HashMap<String, String>>, JanusApiError>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        let window_start = self.current_start;
-        let window_end = window_start.checked_add(self.width)?;
+        let (window_start, window_end) =
+            self.window.resolve_historical_bounds(self.current_evaluation_time)?;
 
-        if window_end > self.evaluation_time {
+        if window_end > self.latest_evaluation_time {
             return None;
         }
 
         // Query storage
-        let events = match self.executor.storage.query(window_start, window_end) {
+        let events = match self.executor.storage.query_half_open(window_start, window_end) {
             Ok(events) => events,
             Err(e) => {
                 return Some(Err(JanusApiError::StorageError(format!("Query failed: {}", e))))
@@ -453,7 +468,8 @@ impl<'a> Iterator for SlidingWindowIterator<'a> {
         let result = self.executor.execute_sparql_on_events(&events, &self.sparql_query);
 
         // Advance window
-        self.current_start += self.slide;
+        self.current_evaluation_time =
+            self.current_evaluation_time.checked_add(self.window.slide)?;
 
         Some(result)
     }

src/parsing/janusql_parser/ast.rs

@@ -74,7 +74,8 @@ impl WindowDefinition {
                     return None;
                 }
 
-                let historical_start = evaluation_time.saturating_sub(offset);
+                // Historical sliding intervals are [T - OFFSET, T - OFFSET + RANGE).
+                let historical_start = evaluation_time.checked_sub(offset)?;
                 let historical_end = historical_start.checked_add(range)?;
                 Some((historical_start, historical_end))
             }

src/parsing/janusql_parser/mod.rs

@@ -346,7 +346,7 @@ impl JanusQLParser {
                     })?;
                     if window.width > offset {
                         return Err(self.parse_error(format!(
-                            "Historical sliding window '{}' has RANGE {} greater than OFFSET {}; the first window would extend beyond the evaluation time",
+                            "Historical sliding window '{}' has RANGE {} greater than OFFSET {}; the historical window would extend beyond the evaluation time",
                             window.window_name, window.width, offset
                         )));
                     }

src/registry/baseline_registry.rs

@@ -73,8 +73,8 @@ mod tests {
             baseline_id: "http://example.org/yesterdayBaseline".to_string(),
             valid_at,
             source_window: "http://example.org/sameMinuteYesterday".to_string(),
-            window_start: 86_340_000,
-            window_end: 86_400_000,
+            window_start: 86_400_000,
+            window_end: 86_460_000,
             variables: vec!["?sensor".to_string(), "?yesterdayAvgValue".to_string()],
             rows: vec![HashMap::from([
                 ("sensor".to_string(), "http://example.org/sensor1".to_string()),

src/storage/segmented_storage/background.rs

@@ -77,55 +77,26 @@ impl StreamingSegmentedStorage {
             events
         };
 
-        let events_ref = &mut events_to_flush;
         let flush_result = (|| -> std::io::Result<()> {
-            let new_segment = Self::write_segment_files(&config, events_ref)?;
-
-            {
-                let mut segments = segments.write().unwrap();
-                segments.push(new_segment);
-                segments.sort_by_key(|s| s.start_timstamp);
-            }
-
+            // The dictionary must be durable before a segment referencing its IDs is committed.
             let dict_path = std::path::Path::new(&config.segment_base_path).join("dictionary.bin");
             let dict = dictionary.read().unwrap();
             dict.save_to_file(&dict_path)?;
 
+            let new_segment = Self::write_segment_files(&config, &mut events_to_flush)?;
+
+            let mut segments = segments.write().unwrap();
+            segments.push(new_segment);
+            segments.sort_by_key(|s| s.start_timstamp);
+
             Ok(())
         })();
 
         if let Err(err) = flush_result {
-            Self::restore_failed_background_flush(&batch_buffer, &events_to_flush);
+            Self::restore_failed_flush(&batch_buffer, &events_to_flush);
             return Err(err);
         }
 
         Ok(())
     }
-
-    fn restore_failed_background_flush(batch_buffer: &Arc<RwLock<BatchBuffer>>, events: &[Event]) {
-        if events.is_empty() {
-            return;
-        }
-
-        let mut buffer = batch_buffer.write().unwrap();
-        for event in events.iter().rev().cloned() {
-            buffer.events.push_front(event);
-            buffer.total_bytes += std::mem::size_of::<Event>();
-        }
-
-        let restored_oldest = events.first().map(|event| event.timestamp);
-        let restored_newest = events.last().map(|event| event.timestamp);
-
-        buffer.oldest_timestamp_bound = match (buffer.oldest_timestamp_bound, restored_oldest) {
-            (Some(existing), Some(restored)) => Some(existing.min(restored)),
-            (None, restored) => restored,
-            (existing, None) => existing,
-        };
-
-        buffer.newest_timestamp_bound = match (buffer.newest_timestamp_bound, restored_newest) {
-            (Some(existing), Some(restored)) => Some(existing.max(restored)),
-            (None, restored) => restored,
-            (existing, None) => existing,
-        };
-    }
 }