From 651507001c0096757c572dcfc860449c529eb147 Mon Sep 17 00:00:00 2001 From: Kush Bisen Date: Thu, 20 Aug 2026 16:25:08 +0200 Subject: [PATCH 1/3] fix: correct historical bounds and segmented storage recovery --- src/api/janus_api/tests.rs | 16 +- src/execution/historical_executor.rs | 62 +++--- src/parsing/janusql_parser/ast.rs | 5 +- src/storage/segmented_storage/background.rs | 45 +---- src/storage/segmented_storage/mod.rs | 31 ++- src/storage/segmented_storage/segment.rs | 179 ++++++++++++++---- .../operators/historical_sliding_window.rs | 22 +-- tests/historical_sliding_window_test.rs | 6 +- tests/historical_window_bounds_test.rs | 19 +- tests/public_spec_behavior_test.rs | 6 +- tests/segmented_storage_error_test.rs | 29 +++ tests/segmented_storage_regression_test.rs | 56 ++++++ 12 files changed, 332 insertions(+), 144 deletions(-) diff --git a/src/api/janus_api/tests.rs b/src/api/janus_api/tests.rs index 884d362..1faed20 100644 --- a/src/api/janus_api/tests.rs +++ b/src/api/janus_api/tests.rs @@ -718,7 +718,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim StreamingSegmentedStorage::new(config).expect("Failed to create segmented storage"), ); - for (timestamp, value) in [(86_400_002, "10"), (86_460_000, "20")] { + for (timestamp, value) in [(86_340_002, "10"), (86_400_000, "20")] { storage .write_rdf( timestamp, @@ -730,7 +730,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim .expect("Failed to write historical RDF event"); } storage.flush().expect("Failed to flush storage"); - for (timestamp, value) in [(86_460_002, "30"), (86_520_000, "50")] { + for (timestamp, value) in [(86_400_002, "30"), (86_460_000, "50")] { storage .write_rdf( timestamp, @@ -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(86_340_001, 86_400_001) .expect("first historical range should query") .len(), 2 ); assert_eq!( storage - .query_rdf(86_460_001, 86_520_001) + .query_rdf(86_400_001, 86_460_001) .expect("second historical range should query") .len(), 2 @@ -910,8 +910,8 @@ HAVING(AVG(?value) > ?yesterdayAvgValue) let second_snapshot = baseline_registry .get_snapshot("http://example.org/yesterdayBaseline", 172_860_001) .expect("expected snapshot at second evaluation time"); - assert_eq!(first_snapshot.window_start, 86_400_001); - assert_eq!(first_snapshot.window_end, 86_460_001); - assert_eq!(second_snapshot.window_start, 86_460_001); - assert_eq!(second_snapshot.window_end, 86_520_001); + assert_eq!(first_snapshot.window_start, 86_340_001); + assert_eq!(first_snapshot.window_end, 86_400_001); + assert_eq!(second_snapshot.window_start, 86_400_001); + assert_eq!(second_snapshot.window_end, 86_460_001); } diff --git a/src/execution/historical_executor.rs b/src/execution/historical_executor.rs index 3320da7..c6a3f58 100644 --- a/src/execution/historical_executor.rs +++ b/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}; @@ -166,23 +165,16 @@ impl HistoricalExecutor { window: &WindowDefinition, sparql_query: &'a str, ) -> impl Iterator>, 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,36 +389,29 @@ 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, } @@ -434,10 +419,10 @@ impl<'a> Iterator for SlidingWindowIterator<'a> { type Item = Result>, JanusApiError>; fn next(&mut self) -> Option { - 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; } @@ -453,7 +438,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) } @@ -557,7 +543,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 4); + assert_eq!(results.len(), 6); assert!(results.iter().all(|result| result.is_ok())); } @@ -586,7 +572,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 2); + assert_eq!(results.len(), 3); assert!(results.iter().all(|result| result.is_ok())); } diff --git a/src/parsing/janusql_parser/ast.rs b/src/parsing/janusql_parser/ast.rs index b597323..db539d9 100644 --- a/src/parsing/janusql_parser/ast.rs +++ b/src/parsing/janusql_parser/ast.rs @@ -74,8 +74,9 @@ impl WindowDefinition { return None; } - let historical_start = evaluation_time.saturating_sub(offset); - let historical_end = historical_start.checked_add(range)?; + // Historical sliding intervals are [T - OFFSET - RANGE, T - OFFSET]. + let historical_end = evaluation_time.checked_sub(offset)?; + let historical_start = historical_end.checked_sub(range)?; Some((historical_start, historical_end)) } } diff --git a/src/storage/segmented_storage/background.rs b/src/storage/segmented_storage/background.rs index 6a7a108..981aae5 100644 --- a/src/storage/segmented_storage/background.rs +++ b/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>, 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::(); - } - - 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, - }; - } } diff --git a/src/storage/segmented_storage/mod.rs b/src/storage/segmented_storage/mod.rs index 49ab2b9..cb6c20c 100644 --- a/src/storage/segmented_storage/mod.rs +++ b/src/storage/segmented_storage/mod.rs @@ -35,14 +35,39 @@ impl StreamingSegmentedStorage { // Load or create dictionary let dict_path = std::path::Path::new(&config.segment_base_path).join("dictionary.bin"); + let has_persisted_segments = std::fs::read_dir(&config.segment_base_path)?.any(|entry| { + entry.ok().is_some_and(|entry| { + entry.file_type().map(|kind| kind.is_file()).unwrap_or(false) + && entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".log")) + }) + }); let dictionary = if dict_path.exists() { match Dictionary::load_from_file(&dict_path) { Ok(dict) => dict, Err(e) => { - eprintln!("Warning: Failed to load dictionary: {}, creating new one", e); + if has_persisted_segments { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "Cannot open persisted segment data without a readable dictionary '{}': {e}", + dict_path.display() + ), + )); + } Dictionary::new() } } + } else if has_persisted_segments { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + format!( + "Cannot open persisted segment data because dictionary '{}' is missing", + dict_path.display() + ), + )); } else { Dictionary::new() }; @@ -158,9 +183,7 @@ impl StreamingSegmentedStorage { /// This is useful when you need to ensure data is persisted immediately. pub fn flush(&self) -> std::io::Result<()> { self.ensure_background_flush_healthy()?; - self.flush_batch_buffer_to_segment()?; - self.save_dictionary()?; - Ok(()) + self.flush_batch_buffer_to_segment() } /// Shutdown the storage system gracefully, ensuring all data is flushed to disk. diff --git a/src/storage/segmented_storage/segment.rs b/src/storage/segmented_storage/segment.rs index 653c096..79bda21 100644 --- a/src/storage/segmented_storage/segment.rs +++ b/src/storage/segmented_storage/segment.rs @@ -1,6 +1,7 @@ use std::{ - io::{BufWriter, Seek, Write}, + io::{BufWriter, Read, Seek, SeekFrom, Write}, sync::atomic::{AtomicU64, Ordering}, + sync::{Arc, RwLock}, time::{SystemTime, UNIX_EPOCH}, }; @@ -9,7 +10,7 @@ use crate::{ encoding::{encode_record, RECORD_SIZE}, Event, }, - storage::util::{EnhancedSegmentMetadata, IndexBlock, StreamingConfig}, + storage::util::{BatchBuffer, EnhancedSegmentMetadata, IndexBlock, StreamingConfig}, }; use super::StreamingSegmentedStorage; @@ -43,18 +44,50 @@ impl StreamingSegmentedStorage { events }; - let segment = Self::write_segment_files(&self.config, &mut events_to_flush)?; + // The dictionary must be durable before a segment referencing its IDs is committed. + let flush_result = (|| -> std::io::Result<()> { + self.save_dictionary()?; + let segment = Self::write_segment_files(&self.config, &mut events_to_flush)?; - { let mut segments = self.segments.write().unwrap(); segments.push(segment); segments.sort_by_key(|s| s.start_timstamp); + Ok(()) + })(); + + if let Err(err) = flush_result { + Self::restore_failed_flush(&self.batch_buffer, &events_to_flush); + return Err(err); } - self.save_dictionary()?; Ok(()) } + pub(super) fn restore_failed_flush(batch_buffer: &Arc>, 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::(); + } + + let restored_oldest = events.iter().map(|event| event.timestamp).min(); + let restored_newest = events.iter().map(|event| event.timestamp).max(); + 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, + }; + } + pub(crate) fn write_segment_files( config: &StreamingConfig, events: &mut [Event], @@ -222,13 +255,25 @@ impl StreamingSegmentedStorage { let index_path = format!("{}/segment-{}.idx", segment_dir, segment_id); if let Ok(_metadata) = fs::metadata(&data_path) { - let (index_directory, start_ts, end_ts, record_count) = - if fs::metadata(&index_path).is_ok() { - Self::load_index_directory_from_file(&index_path) - .unwrap_or_else(|_| (Vec::new(), 0, u64::MAX, 0)) - } else { - (Vec::new(), 0, u64::MAX, 0) - }; + let (start_ts, end_ts, record_count) = + Self::load_segment_log_metadata(&data_path)?; + let mut index_directory = if fs::metadata(&index_path).is_ok() { + Self::load_index_directory_from_file( + &index_path, + self.config.entries_per_index_block, + )? + } else { + Vec::new() + }; + + for block_index in 0..index_directory.len() { + index_directory[block_index].max_timestamp = + if block_index + 1 < index_directory.len() { + index_directory[block_index + 1].min_timestamp + } else { + end_ts + }; + } let segment = EnhancedSegmentMetadata { start_timstamp: start_ts, @@ -256,51 +301,73 @@ impl StreamingSegmentedStorage { Ok(()) } + fn load_segment_log_metadata(data_path: &str) -> std::io::Result<(u64, u64, u64)> { + let mut file = std::fs::File::open(data_path)?; + let byte_len = file.metadata()?.len(); + if byte_len == 0 || byte_len % RECORD_SIZE as u64 != 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("Segment log '{data_path}' is empty or truncated"), + )); + } + + let record_count = byte_len / RECORD_SIZE as u64; + let mut record = [0u8; RECORD_SIZE]; + file.read_exact(&mut record)?; + let (start_timestamp, ..) = crate::core::encoding::decode_record(&record); + file.seek(SeekFrom::Start((record_count - 1) * RECORD_SIZE as u64))?; + file.read_exact(&mut record)?; + let (end_timestamp, ..) = crate::core::encoding::decode_record(&record); + Ok((start_timestamp, end_timestamp, record_count)) + } + pub(super) fn load_index_directory_from_file( index_path: &str, - ) -> std::io::Result<(Vec, u64, u64, u64)> { - use std::io::Read; - + entries_per_index_block: usize, + ) -> std::io::Result> { + const INDEX_ENTRY_SIZE: usize = 16; + if entries_per_index_block == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "entries_per_index_block must be greater than zero", + )); + } let mut file = std::fs::File::open(index_path)?; let mut buffer = Vec::new(); file.read_to_end(&mut buffer)?; if buffer.is_empty() { - return Ok((Vec::new(), 0, u64::MAX, 0)); + return Ok(Vec::new()); + } + if buffer.len() % INDEX_ENTRY_SIZE != 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("Sparse index '{index_path}' is truncated"), + )); } let mut index_directory = Vec::new(); let mut file_offset = 0u64; - let mut global_min_ts = u64::MAX; - let mut global_max_ts = 0u64; - let mut total_records = 0u64; - - let entries_per_block = 1000; let mut current_block_start = 0; while current_block_start < buffer.len() { - let block_size = - std::cmp::min(entries_per_block * 16, buffer.len() - current_block_start); + let block_size = std::cmp::min( + entries_per_index_block * INDEX_ENTRY_SIZE, + buffer.len() - current_block_start, + ); let block_end = current_block_start + block_size; let block_entries = block_end - current_block_start; - let entry_count = (block_entries / 16) as u32; - - if entry_count == 0 { - break; - } + let entry_count = (block_entries / INDEX_ENTRY_SIZE) as u32; let first_ts = u64::from_le_bytes( buffer[current_block_start..current_block_start + 8].try_into().unwrap(), ); - let last_entry_start = current_block_start + ((entry_count - 1) as usize * 16); + let last_entry_start = + current_block_start + ((entry_count - 1) as usize * INDEX_ENTRY_SIZE); let last_ts = u64::from_le_bytes( buffer[last_entry_start..last_entry_start + 8].try_into().unwrap(), ); - global_min_ts = global_min_ts.min(first_ts); - global_max_ts = global_max_ts.max(last_ts); - total_records += entry_count as u64; - index_directory.push(IndexBlock { min_timestamp: first_ts, max_timestamp: last_ts, @@ -312,6 +379,50 @@ impl StreamingSegmentedStorage { current_block_start = block_end; } - Ok((index_directory, global_min_ts, global_max_ts, total_records)) + Ok(index_directory) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn recovery_uses_log_metadata_and_configured_index_block_size() { + let temp_dir = TempDir::new().unwrap(); + let config = StreamingConfig { + segment_base_path: temp_dir.path().to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 2, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + for timestamp in (10..=70).step_by(10) { + storage + .write(Event { timestamp, subject: 0, predicate: 0, object: 0, graph: 0 }) + .unwrap(); + } + storage.flush().unwrap(); + } + + let storage = StreamingSegmentedStorage::new(config).unwrap(); + let segments = storage.segments.read().unwrap(); + assert_eq!(segments.len(), 1); + let segment = &segments[0]; + assert_eq!(segment.record_count, 7); + assert_eq!(segment.start_timstamp, 10); + assert_eq!(segment.end_timestamp, 70); + assert_eq!(segment.index_directory.len(), 2); + assert_eq!(segment.index_directory.last().unwrap().max_timestamp, 70); + drop(segments); + + let tail = storage.query(70, 70).unwrap(); + assert_eq!(tail.len(), 1); + assert_eq!(tail[0].timestamp, 70); } } diff --git a/src/stream/operators/historical_sliding_window.rs b/src/stream/operators/historical_sliding_window.rs index 5a0a898..dcb3e22 100644 --- a/src/stream/operators/historical_sliding_window.rs +++ b/src/stream/operators/historical_sliding_window.rs @@ -8,8 +8,8 @@ use std::rc::Rc; pub struct HistoricalSlidingWindowOperator { storage: Rc, window_def: WindowDefinition, - current_start: u64, - evaluation_time: u64, + current_evaluation_time: u64, + latest_evaluation_time: u64, } impl HistoricalSlidingWindowOperator { @@ -25,16 +25,11 @@ impl HistoricalSlidingWindowOperator { .unwrap() .as_millis() as u64; - // Offset is mandatory for HistoricalSliding windows as per the parser and requirements. - // We subtract it from the query_start to "go back" in time. - let offset = window_def.offset.expect("Offset must be defined for HistoricalSlidingWindow"); - let start_time = now.saturating_sub(offset); - HistoricalSlidingWindowOperator { storage, window_def, - current_start: start_time, - evaluation_time: now, + current_evaluation_time: now, + latest_evaluation_time: now, } } } @@ -43,10 +38,10 @@ impl Iterator for HistoricalSlidingWindowOperator { type Item = Vec; fn next(&mut self) -> Option { - let window_start = self.current_start; - let window_end = window_start.checked_add(self.window_def.width)?; + let (window_start, window_end) = + self.window_def.resolve_historical_bounds(self.current_evaluation_time)?; - if window_end > self.evaluation_time { + if window_end > self.latest_evaluation_time { return None; } @@ -55,7 +50,8 @@ impl Iterator for HistoricalSlidingWindowOperator { match events_result { Ok(events) => { // Advance the window - self.current_start += self.window_def.slide; + self.current_evaluation_time = + self.current_evaluation_time.checked_add(self.window_def.slide)?; Some(events) } Err(e) => { diff --git a/tests/historical_sliding_window_test.rs b/tests/historical_sliding_window_test.rs index a473be3..37c264c 100644 --- a/tests/historical_sliding_window_test.rs +++ b/tests/historical_sliding_window_test.rs @@ -77,7 +77,7 @@ fn test_historical_sliding_window_with_real_iris() { let mut operator = HistoricalSlidingWindowOperator::new(storage.clone(), window_def); - // Window 1: [now-500, now-300] + // Window 1: [now-700, now-500] let w1 = operator.next().unwrap(); assert!(w1.len() >= 2); // At least 2 events (type + value for first sensor) @@ -90,11 +90,11 @@ fn test_historical_sliding_window_with_real_iris() { first_event.timestamp ); - // Window 2: [now-400, now-200] + // Window 2: [now-600, now-400] let w2 = operator.next().unwrap(); assert!(w2.len() >= 2); - // Window 3: [now-300, now-100] + // Window 3: [now-500, now-300] let w3 = operator.next().unwrap(); assert!(w3.len() >= 2); } diff --git a/tests/historical_window_bounds_test.rs b/tests/historical_window_bounds_test.rs index 017c7fc..88eb5ff 100644 --- a/tests/historical_window_bounds_test.rs +++ b/tests/historical_window_bounds_test.rs @@ -31,13 +31,13 @@ fn fixed_window() -> WindowDefinition { #[test] fn resolves_sliding_historical_bounds_for_first_evaluation() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_400_000, 86_460_000))); + assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_340_000, 86_400_000))); } #[test] fn resolves_sliding_historical_bounds_for_next_evaluation() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_460_000, 86_520_000))); + assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_400_000, 86_460_000))); } #[test] @@ -47,6 +47,21 @@ fn sliding_historical_bounds_return_none_when_first_window_would_cross_evaluatio assert_eq!(window.resolve_historical_bounds(172_800_000), None); } +#[test] +fn sliding_historical_bounds_return_none_on_underflow() { + let window = sliding_window(); + assert_eq!(window.resolve_historical_bounds(86_400_000), None); +} + +#[test] +fn sliding_historical_bounds_have_the_configured_width() { + let window = sliding_window(); + let (start, end) = window.resolve_historical_bounds(172_800_000).unwrap(); + assert_eq!(end, 172_800_000 - 86_400_000); + assert_eq!(start, end - 60_000); + assert_eq!(end - start, 60_000); +} + #[test] fn resolves_fixed_historical_bounds_independent_of_evaluation_time() { let window = fixed_window(); diff --git a/tests/public_spec_behavior_test.rs b/tests/public_spec_behavior_test.rs index 8fc6e6f..9bcf5ab 100644 --- a/tests/public_spec_behavior_test.rs +++ b/tests/public_spec_behavior_test.rs @@ -327,7 +327,7 @@ fn spec_hybrid_historical_sliding_query_parses() { } #[test] -fn spec_historical_sliding_bounds_follow_t_minus_offset_plus_range_formula() { +fn spec_historical_sliding_bounds_follow_t_minus_offset_minus_range_formula() { let window = WindowDefinition { window_name: "http://example.org/previousHour".to_string(), source_kind: SourceKind::Log, @@ -341,8 +341,8 @@ fn spec_historical_sliding_bounds_follow_t_minus_offset_plus_range_formula() { }; let evaluation_time = 200_000_000; - let expected_start = evaluation_time - 86_400_000; - let expected_end = expected_start + 3_600_000; + let expected_end = evaluation_time - 86_400_000; + let expected_start = expected_end - 3_600_000; assert_eq!( window.resolve_historical_bounds(evaluation_time), diff --git a/tests/segmented_storage_error_test.rs b/tests/segmented_storage_error_test.rs index bbde812..0a70388 100644 --- a/tests/segmented_storage_error_test.rs +++ b/tests/segmented_storage_error_test.rs @@ -50,3 +50,32 @@ fn test_background_flush_failure_surfaces_as_storage_error() { "unexpected shutdown error: {shutdown_err}" ); } + +#[test] +fn synchronous_flush_does_not_commit_a_segment_when_dictionary_persistence_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let storage = StreamingSegmentedStorage::new(StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1024 * 1024, + sparse_interval: 1, + entries_per_index_block: 2, + }) + .unwrap(); + + fs::create_dir(storage_dir.join("dictionary.bin")).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + assert!(storage.flush().is_err()); + assert!(fs::read_dir(&storage_dir).unwrap().all(|entry| !entry + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".log"))); + assert_eq!(storage.query(0, 2_000).unwrap().len(), 1); + + fs::remove_dir(storage_dir.join("dictionary.bin")).unwrap(); + storage.flush().unwrap(); + assert_eq!(storage.query(0, 2_000).unwrap().len(), 1); +} diff --git a/tests/segmented_storage_regression_test.rs b/tests/segmented_storage_regression_test.rs index a189480..55f707f 100644 --- a/tests/segmented_storage_regression_test.rs +++ b/tests/segmented_storage_regression_test.rs @@ -270,3 +270,59 @@ fn test_shutdown_race_safety() { } } } + +#[test] +fn opening_persisted_segments_without_dictionary_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let config = StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 1, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + storage.flush().unwrap(); + } + + fs::remove_file(storage_dir.join("dictionary.bin")).unwrap(); + let err = match StreamingSegmentedStorage::new(config) { + Ok(_) => panic!("missing dictionary must fail"), + Err(err) => err, + }; + assert!(err.to_string().contains("persisted segment data")); + assert!(err.to_string().contains("dictionary")); +} + +#[test] +fn opening_persisted_segments_with_corrupt_dictionary_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let config = StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 1, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + storage.flush().unwrap(); + } + + fs::write(storage_dir.join("dictionary.bin"), b"not a dictionary").unwrap(); + let err = match StreamingSegmentedStorage::new(config) { + Ok(_) => panic!("corrupt dictionary must fail"), + Err(err) => err, + }; + assert!(err.to_string().contains("persisted segment data")); + assert!(err.to_string().contains("readable dictionary")); +} From ec616ba3dd7204e050f09b221e7bc9b107fc823a Mon Sep 17 00:00:00 2001 From: Kush Bisen Date: Thu, 10 Sep 2026 16:00:44 +0200 Subject: [PATCH 2/3] Fix historical sliding window semantics --- benches/historical_sliding.rs | 2 +- docs/JANUSQL.md | 8 ++- docs/WINDOW_TYPES_EXPLAINED.md | 6 +- src/api/janus_api/baseline.rs | 18 ++++-- src/api/janus_api/tests.rs | 16 ++--- src/bin/hybrid_scaling_combined.rs | 7 +- src/execution/historical_executor.rs | 42 ++++++++++-- src/parsing/janusql_parser/ast.rs | 6 +- src/parsing/janusql_parser/mod.rs | 2 +- src/registry/baseline_registry.rs | 4 +- src/storage/segmented_storage/query.rs | 28 ++++++++ .../operators/historical_sliding_window.rs | 2 +- tests/historical_sliding_window_test.rs | 6 +- tests/historical_window_bounds_test.rs | 64 ++++++++++++++++--- tests/janus_api_integration_test.rs | 4 +- tests/janusql_parser_test.rs | 4 +- tests/public_spec_behavior_test.rs | 10 +-- 17 files changed, 178 insertions(+), 51 deletions(-) diff --git a/benches/historical_sliding.rs b/benches/historical_sliding.rs index a4f6afd..68dbbb0 100644 --- a/benches/historical_sliding.rs +++ b/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; diff --git a/docs/JANUSQL.md b/docs/JANUSQL.md index 5c680d3..5ce4b82 100644 --- a/docs/JANUSQL.md +++ b/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 diff --git a/docs/WINDOW_TYPES_EXPLAINED.md b/docs/WINDOW_TYPES_EXPLAINED.md index ee9d9b2..24e8025 100644 --- a/docs/WINDOW_TYPES_EXPLAINED.md +++ b/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 diff --git a/src/api/janus_api/baseline.rs b/src/api/janus_api/baseline.rs index 21b36f5..c5f1e13 100644 --- a/src/api/janus_api/baseline.rs +++ b/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, diff --git a/src/api/janus_api/tests.rs b/src/api/janus_api/tests.rs index 1faed20..9522b54 100644 --- a/src/api/janus_api/tests.rs +++ b/src/api/janus_api/tests.rs @@ -718,7 +718,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim StreamingSegmentedStorage::new(config).expect("Failed to create segmented storage"), ); - for (timestamp, value) in [(86_340_002, "10"), (86_400_000, "20")] { + for (timestamp, value) in [(86_400_002, "10"), (86_460_000, "20")] { storage .write_rdf( timestamp, @@ -730,7 +730,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim .expect("Failed to write historical RDF event"); } storage.flush().expect("Failed to flush storage"); - for (timestamp, value) in [(86_400_002, "30"), (86_460_000, "50")] { + for (timestamp, value) in [(86_460_002, "30"), (86_520_000, "50")] { storage .write_rdf( timestamp, @@ -784,14 +784,14 @@ HAVING(AVG(?value) > ?yesterdayAvgValue) let latest_rows = Arc::new(RwLock::new(HashMap::new())); assert_eq!( storage - .query_rdf(86_340_001, 86_400_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_400_001, 86_460_001) + .query_rdf_half_open(86_460_001, 86_520_001) .expect("second historical range should query") .len(), 2 @@ -910,8 +910,8 @@ HAVING(AVG(?value) > ?yesterdayAvgValue) let second_snapshot = baseline_registry .get_snapshot("http://example.org/yesterdayBaseline", 172_860_001) .expect("expected snapshot at second evaluation time"); - assert_eq!(first_snapshot.window_start, 86_340_001); - assert_eq!(first_snapshot.window_end, 86_400_001); - assert_eq!(second_snapshot.window_start, 86_400_001); - assert_eq!(second_snapshot.window_end, 86_460_001); + assert_eq!(first_snapshot.window_start, 86_400_001); + assert_eq!(first_snapshot.window_end, 86_460_001); + assert_eq!(second_snapshot.window_start, 86_460_001); + assert_eq!(second_snapshot.window_end, 86_520_001); } diff --git a/src/bin/hybrid_scaling_combined.rs b/src/bin/hybrid_scaling_combined.rs index e442dcc..d2df6d0 100644 --- a/src/bin/hybrid_scaling_combined.rs +++ b/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)) diff --git a/src/execution/historical_executor.rs b/src/execution/historical_executor.rs index c6a3f58..bed26ec 100644 --- a/src/execution/historical_executor.rs +++ b/src/execution/historical_executor.rs @@ -102,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>, 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( @@ -121,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( @@ -427,7 +457,7 @@ impl<'a> Iterator for SlidingWindowIterator<'a> { } // 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)))) @@ -543,7 +573,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 6); + assert_eq!(results.len(), 4); assert!(results.iter().all(|result| result.is_ok())); } @@ -572,7 +602,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 3); + assert_eq!(results.len(), 2); assert!(results.iter().all(|result| result.is_ok())); } diff --git a/src/parsing/janusql_parser/ast.rs b/src/parsing/janusql_parser/ast.rs index db539d9..01fb7b3 100644 --- a/src/parsing/janusql_parser/ast.rs +++ b/src/parsing/janusql_parser/ast.rs @@ -74,9 +74,9 @@ impl WindowDefinition { return None; } - // Historical sliding intervals are [T - OFFSET - RANGE, T - OFFSET]. - let historical_end = evaluation_time.checked_sub(offset)?; - let historical_start = historical_end.checked_sub(range)?; + // 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)) } } diff --git a/src/parsing/janusql_parser/mod.rs b/src/parsing/janusql_parser/mod.rs index 4a5ee23..21174d1 100644 --- a/src/parsing/janusql_parser/mod.rs +++ b/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 ))); } diff --git a/src/registry/baseline_registry.rs b/src/registry/baseline_registry.rs index c4ceab2..975875d 100644 --- a/src/registry/baseline_registry.rs +++ b/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()), diff --git a/src/storage/segmented_storage/query.rs b/src/storage/segmented_storage/query.rs index 4e389bf..575060d 100644 --- a/src/storage/segmented_storage/query.rs +++ b/src/storage/segmented_storage/query.rs @@ -11,6 +11,23 @@ use crate::{ use super::StreamingSegmentedStorage; impl StreamingSegmentedStorage { + /// Query events in the half-open timestamp interval `[start, end)`. + /// + /// The underlying storage query API is inclusive at both ends for + /// compatibility with fixed historical ranges and point lookups. Historical + /// sliding windows use this adapter so their resolved end remains exclusive. + pub fn query_half_open( + &self, + start_timestamp: u64, + end_timestamp: u64, + ) -> std::io::Result> { + self.ensure_background_flush_healthy()?; + if start_timestamp >= end_timestamp { + return Ok(Vec::new()); + } + self.query(start_timestamp, end_timestamp - 1) + } + /// Query events within a timestamp range from the storage system but result in encoded Events and not RDFEvents. pub fn query(&self, start_timestamp: u64, end_timestamp: u64) -> std::io::Result> { self.ensure_background_flush_healthy()?; @@ -54,6 +71,17 @@ impl StreamingSegmentedStorage { Ok(encoded_events.into_iter().map(|event| event.decode(&dict)).collect()) } + /// Query RDF events in the half-open timestamp interval `[start, end)`. + pub fn query_rdf_half_open( + &self, + start_timestamp: u64, + end_timestamp: u64, + ) -> std::io::Result> { + let encoded_events = self.query_half_open(start_timestamp, end_timestamp)?; + let dict = self.dictionary.read().unwrap(); + Ok(encoded_events.into_iter().map(|event| event.decode(&dict)).collect()) + } + // Query a segment using two-level indexing fn query_segment_two_level( &self, diff --git a/src/stream/operators/historical_sliding_window.rs b/src/stream/operators/historical_sliding_window.rs index dcb3e22..00bee81 100644 --- a/src/stream/operators/historical_sliding_window.rs +++ b/src/stream/operators/historical_sliding_window.rs @@ -45,7 +45,7 @@ impl Iterator for HistoricalSlidingWindowOperator { return None; } - let events_result = self.storage.query(window_start, window_end); + let events_result = self.storage.query_half_open(window_start, window_end); match events_result { Ok(events) => { diff --git a/tests/historical_sliding_window_test.rs b/tests/historical_sliding_window_test.rs index 37c264c..4d9afb2 100644 --- a/tests/historical_sliding_window_test.rs +++ b/tests/historical_sliding_window_test.rs @@ -77,7 +77,7 @@ fn test_historical_sliding_window_with_real_iris() { let mut operator = HistoricalSlidingWindowOperator::new(storage.clone(), window_def); - // Window 1: [now-700, now-500] + // Window 1: [now-500, now-300) let w1 = operator.next().unwrap(); assert!(w1.len() >= 2); // At least 2 events (type + value for first sensor) @@ -90,11 +90,11 @@ fn test_historical_sliding_window_with_real_iris() { first_event.timestamp ); - // Window 2: [now-600, now-400] + // Window 2: [now-400, now-200) let w2 = operator.next().unwrap(); assert!(w2.len() >= 2); - // Window 3: [now-500, now-300] + // Window 3: [now-300, now-100) let w3 = operator.next().unwrap(); assert!(w3.len() >= 2); } diff --git a/tests/historical_window_bounds_test.rs b/tests/historical_window_bounds_test.rs index 88eb5ff..8c75310 100644 --- a/tests/historical_window_bounds_test.rs +++ b/tests/historical_window_bounds_test.rs @@ -1,4 +1,7 @@ use janus::parsing::janusql_parser::{SourceKind, WindowDefinition, WindowType}; +use janus::storage::segmented_storage::StreamingSegmentedStorage; +use janus::storage::util::StreamingConfig; +use tempfile::TempDir; fn sliding_window() -> WindowDefinition { WindowDefinition { @@ -29,39 +32,82 @@ fn fixed_window() -> WindowDefinition { } #[test] -fn resolves_sliding_historical_bounds_for_first_evaluation() { +fn resolves_sliding_historical_bounds_with_range_less_than_offset() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_340_000, 86_400_000))); + assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_400_000, 86_460_000))); } #[test] fn resolves_sliding_historical_bounds_for_next_evaluation() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_400_000, 86_460_000))); + assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_460_000, 86_520_000))); } #[test] -fn sliding_historical_bounds_return_none_when_first_window_would_cross_evaluation_time() { +fn sliding_historical_bounds_reject_range_greater_than_offset() { let mut window = sliding_window(); - window.width = 90_000_000; + window.width = 86_400_001; assert_eq!(window.resolve_historical_bounds(172_800_000), None); } #[test] -fn sliding_historical_bounds_return_none_on_underflow() { +fn sliding_historical_bounds_return_none_when_evaluation_precedes_offset() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(86_400_000), None); + assert_eq!(window.resolve_historical_bounds(86_399_999), None); } #[test] fn sliding_historical_bounds_have_the_configured_width() { let window = sliding_window(); let (start, end) = window.resolve_historical_bounds(172_800_000).unwrap(); - assert_eq!(end, 172_800_000 - 86_400_000); - assert_eq!(start, end - 60_000); + assert_eq!(start, 172_800_000 - 86_400_000); + assert_eq!(end, start + 60_000); assert_eq!(end - start, 60_000); } +#[test] +fn sliding_historical_bounds_end_at_evaluation_time_when_range_equals_offset() { + let mut window = sliding_window(); + window.width = 86_400_000; + + assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_400_000, 172_800_000))); +} + +#[test] +fn sliding_historical_storage_query_is_half_open() { + let temp_dir = TempDir::new().expect("failed to create temporary storage directory"); + let storage = StreamingSegmentedStorage::new(StreamingConfig { + segment_base_path: temp_dir.path().to_string_lossy().into_owned(), + ..StreamingConfig::default() + }) + .expect("failed to create storage"); + + for timestamp in [100, 150] { + storage + .write_rdf( + timestamp, + "http://example.org/sensor", + "http://example.org/value", + ×tamp.to_string(), + "http://example.org/graph", + ) + .expect("failed to write event"); + } + storage.flush().expect("failed to flush storage"); + + let mut window = sliding_window(); + window.width = 50; + window.offset = Some(100); + let (start, end) = window.resolve_historical_bounds(200).expect("bounds should resolve"); + assert_eq!((start, end), (100, 150)); + + let events = storage + .query_rdf_half_open(start, end) + .expect("half-open historical query should succeed"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].timestamp, start); +} + #[test] fn resolves_fixed_historical_bounds_independent_of_evaluation_time() { let window = fixed_window(); diff --git a/tests/janus_api_integration_test.rs b/tests/janus_api_integration_test.rs index 4a0c8a1..9fcdec4 100644 --- a/tests/janus_api_integration_test.rs +++ b/tests/janus_api_integration_test.rs @@ -135,7 +135,9 @@ fn test_register_rejects_historical_sliding_window_when_range_exceeds_offset() { .register_query("invalid_hist_sliding".into(), janusql) .expect_err("historical sliding window should be rejected during registration"); - assert!(err.to_string().contains("first window would extend beyond the evaluation time")); + assert!(err + .to_string() + .contains("the historical window would extend beyond the evaluation time")); } #[test] diff --git a/tests/janusql_parser_test.rs b/tests/janusql_parser_test.rs index ba4bd78..6b90735 100644 --- a/tests/janusql_parser_test.rs +++ b/tests/janusql_parser_test.rs @@ -487,7 +487,9 @@ fn test_sliding_historical_log_window_rejects_range_greater_than_offset() { "#; let err = parser.parse(query).expect_err("range greater than offset should be rejected"); - assert!(err.to_string().contains("first window would extend beyond the evaluation time")); + assert!(err + .to_string() + .contains("the historical window would extend beyond the evaluation time")); } #[test] diff --git a/tests/public_spec_behavior_test.rs b/tests/public_spec_behavior_test.rs index 9bcf5ab..823bf83 100644 --- a/tests/public_spec_behavior_test.rs +++ b/tests/public_spec_behavior_test.rs @@ -327,7 +327,7 @@ fn spec_hybrid_historical_sliding_query_parses() { } #[test] -fn spec_historical_sliding_bounds_follow_t_minus_offset_minus_range_formula() { +fn spec_historical_sliding_bounds_follow_offset_then_range_formula() { let window = WindowDefinition { window_name: "http://example.org/previousHour".to_string(), source_kind: SourceKind::Log, @@ -341,8 +341,8 @@ fn spec_historical_sliding_bounds_follow_t_minus_offset_minus_range_formula() { }; let evaluation_time = 200_000_000; - let expected_end = evaluation_time - 86_400_000; - let expected_start = expected_end - 3_600_000; + let expected_start = evaluation_time - 86_400_000; + let expected_end = expected_start + 3_600_000; assert_eq!( window.resolve_historical_bounds(evaluation_time), @@ -371,7 +371,9 @@ fn spec_historical_sliding_log_window_rejects_range_greater_than_offset() { .parse(&invalid_query) .expect_err("historical sliding log window should reject RANGE > OFFSET"); - assert!(err.to_string().contains("first window would extend beyond the evaluation time")); + assert!(err + .to_string() + .contains("the historical window would extend beyond the evaluation time")); } #[test] From 189b234d11f2a330247772384b80b921c895d27d Mon Sep 17 00:00:00 2001 From: Kush Bisen Date: Thu, 10 Sep 2026 16:13:56 +0200 Subject: [PATCH 3/3] Fix Clippy warnings in storage recovery --- src/storage/segmented_storage/mod.rs | 2 +- src/storage/segmented_storage/query.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/storage/segmented_storage/mod.rs b/src/storage/segmented_storage/mod.rs index cb6c20c..7297264 100644 --- a/src/storage/segmented_storage/mod.rs +++ b/src/storage/segmented_storage/mod.rs @@ -36,7 +36,7 @@ impl StreamingSegmentedStorage { // Load or create dictionary let dict_path = std::path::Path::new(&config.segment_base_path).join("dictionary.bin"); let has_persisted_segments = std::fs::read_dir(&config.segment_base_path)?.any(|entry| { - entry.ok().is_some_and(|entry| { + entry.is_ok_and(|entry| { entry.file_type().map(|kind| kind.is_file()).unwrap_or(false) && entry .file_name() diff --git a/src/storage/segmented_storage/query.rs b/src/storage/segmented_storage/query.rs index 575060d..03fd66a 100644 --- a/src/storage/segmented_storage/query.rs +++ b/src/storage/segmented_storage/query.rs @@ -151,7 +151,7 @@ impl StreamingSegmentedStorage { let mut buffer = vec![0u8; block_size]; index_file.read_exact(&mut buffer)?; - for chunk in buffer.chunks_exact(16) { + for chunk in buffer.chunks(16) { let timestamp = u64::from_le_bytes(chunk[0..8].try_into().unwrap()); let offset = u64::from_be_bytes(chunk[8..16].try_into().unwrap()); sparse_entries.push((timestamp, offset));