Skip to content

fix: correct historical bounds and segmented storage recovery #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
argahsuknesib wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: correct historical bounds and segmented storage recovery #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions src/api/janus_api/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}
62 changes: 24 additions & 38 deletions src/execution/historical_executor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -166,23 +165,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(),
}
}
Expand Down Expand Up @@ -397,47 +389,40 @@ 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;
}

Expand All @@ -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)
}
Expand Down Expand Up @@ -557,7 +543,7 @@ mod tests {
.execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }")
.collect::<Vec<_>>();

assert_eq!(results.len(), 4);
assert_eq!(results.len(), 6);
assert!(results.iter().all(|result| result.is_ok()));
}

Expand Down Expand Up @@ -586,7 +572,7 @@ mod tests {
.execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }")
.collect::<Vec<_>>();

assert_eq!(results.len(), 2);
assert_eq!(results.len(), 3);
assert!(results.iter().all(|result| result.is_ok()));
}

Expand Down
5 changes: 3 additions & 2 deletions src/parsing/janusql_parser/ast.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
}
Expand Down
45 changes: 8 additions & 37 deletions src/storage/segmented_storage/background.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
}
}
31 changes: 27 additions & 4 deletions src/storage/segmented_storage/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
};
Expand Down Expand Up @@ -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.
Expand Down
Loading
Loading