Merge pull request #10 from SolidLabResearch/codex/on-log-extension-functions

Add ON LOG parsing and historical extension functions

Author: argahsuknesib

benches/historical_fixed.rs

@@ -1,7 +1,7 @@
 use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
 use janus::{
     execution::historical_executor::HistoricalExecutor,
-    parsing::janusql_parser::{WindowDefinition, WindowType},
+    parsing::janusql_parser::{SourceKind, WindowDefinition, WindowType},
     querying::oxigraph_adapter::OxigraphAdapter,
     storage::{segmented_storage::StreamingSegmentedStorage, util::StreamingConfig},
 };
@@ -43,6 +43,7 @@ fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
     }
     let window = WindowDefinition {
         window_name: "w".to_string(),
+        source_kind: SourceKind::Stream,
         stream_name: "http://example.org/stream".to_string(),
         width: n as u64,
         slide: n as u64,

benches/historical_sliding.rs

@@ -1,7 +1,7 @@
 use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
 use janus::{
     execution::historical_executor::HistoricalExecutor,
-    parsing::janusql_parser::{WindowDefinition, WindowType},
+    parsing::janusql_parser::{SourceKind, WindowDefinition, WindowType},
     querying::oxigraph_adapter::OxigraphAdapter,
     storage::{segmented_storage::StreamingSegmentedStorage, util::StreamingConfig},
 };
@@ -53,6 +53,7 @@ fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
     }
     let window = WindowDefinition {
         window_name: "w".to_string(),
+        source_kind: SourceKind::Stream,
         stream_name: "http://example.org/stream".to_string(),
         width: RANGE_MS,
         slide: SLIDE_MS,

src/execution/historical_executor.rs

@@ -501,6 +501,7 @@ mod tests {
 
         let window = WindowDefinition {
             window_name: "test_window".to_string(),
+            source_kind: crate::parsing::janusql_parser::SourceKind::Stream,
             stream_name: "test_stream".to_string(),
             width: 1000,
             slide: 100,
@@ -528,6 +529,7 @@ mod tests {
 
         let window = WindowDefinition {
             window_name: "test_window".to_string(),
+            source_kind: crate::parsing::janusql_parser::SourceKind::Stream,
             stream_name: "test_stream".to_string(),
             width: 1000,
             slide: 100,

src/extensions/math.rs

@@ -0,0 +1,56 @@
+//! Pure numeric helpers used by Janus extension functions.
+
+/// Absolute difference between live and historical values: `|live - hist|`.
+pub fn abs_diff(live: f64, hist: f64) -> f64 {
+    (live - hist).abs()
+}
+
+/// Relative change from historical to live: `(live - hist) / hist`.
+///
+/// Returns `f64::NAN` when `hist` is zero.
+pub fn relative_change(live: f64, hist: f64) -> f64 {
+    (live - hist) / hist
+}
+
+/// Z-score of `value` given a distribution with `mean` and `sigma`.
+///
+/// Returns `0.0` when `sigma` is zero.
+pub fn zscore(value: f64, mean: f64, sigma: f64) -> f64 {
+    if sigma.abs() < f64::EPSILON {
+        0.0
+    } else {
+        (value - mean) / sigma
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn abs_diff_positive_delta() {
+        assert!((abs_diff(3.0, 1.0) - 2.0).abs() < f64::EPSILON);
+    }
+
+    #[test]
+    fn abs_diff_negative_delta() {
+        assert!((abs_diff(1.0, 3.0) - 2.0).abs() < f64::EPSILON);
+    }
+
+    #[test]
+    fn relative_change_increase() {
+        let rc = relative_change(1.1, 1.0);
+        assert!((rc - 0.1).abs() < 1e-10);
+    }
+
+    #[test]
+    fn relative_change_zero_hist_is_nan_or_infinite() {
+        let rc = relative_change(1.0, 0.0);
+        assert!(rc.is_nan() || rc.is_infinite());
+    }
+
+    #[test]
+    fn zscore_zero_sigma_returns_zero() {
+        assert!(zscore(99.0, 1.0, 0.0).abs() < f64::EPSILON);
+    }
+}

src/extensions/mod.rs

@@ -0,0 +1,10 @@
+//! Janus SPARQL extension functions.
+//!
+//! This module contains reusable function registry and Oxigraph wiring for
+//! Janus-specific extension functions exposed under the `https://janus.rs/fn#`
+//! namespace.
+
+pub mod math;
+pub mod query_options;
+pub mod registry;
+pub mod rules;

src/extensions/query_options.rs

@@ -0,0 +1,149 @@
+//! Oxigraph evaluator builder preloaded with Janus extension functions.
+
+use oxigraph::model::{Literal, NamedNode, Term};
+use oxigraph::sparql::SparqlEvaluator;
+
+use crate::extensions::math::{abs_diff, relative_change, zscore};
+use crate::extensions::registry::{
+    FunctionRegistry, FN_ABSOLUTE_THRESHOLD, FN_ABS_DIFF, FN_CATCH_UP, FN_IS_OUTLIER,
+    FN_RELATIVE_CHANGE, FN_RELATIVE_THRESHOLD, FN_TREND_DIVERGENT, FN_VOLATILITY_INCREASE,
+    FN_ZSCORE,
+};
+
+fn term_to_f64(term: &Term) -> Option<f64> {
+    if let Term::Literal(literal) = term {
+        literal.value().parse::<f64>().ok()
+    } else {
+        None
+    }
+}
+
+fn bool_term(value: bool) -> Term {
+    Term::Literal(Literal::new_typed_literal(
+        if value { "true" } else { "false" },
+        NamedNode::new("http://www.w3.org/2001/XMLSchema#boolean").unwrap(),
+    ))
+}
+
+fn decimal_term(value: f64) -> Term {
+    Term::Literal(Literal::new_typed_literal(
+        &value.to_string(),
+        NamedNode::new("http://www.w3.org/2001/XMLSchema#decimal").unwrap(),
+    ))
+}
+
+fn args_to_floats(args: &[Term]) -> Option<Vec<f64>> {
+    args.iter().map(term_to_f64).collect()
+}
+
+/// Build a `SparqlEvaluator` with Janus extension functions registered.
+pub fn build_evaluator() -> SparqlEvaluator {
+    let registry = FunctionRegistry::new();
+    let mut evaluator = SparqlEvaluator::new();
+
+    for (uri, rule) in registry.all_rules() {
+        let name = NamedNode::new(uri).expect("constant URI must be valid");
+        evaluator = evaluator.with_custom_function(name, move |args| {
+            let floats = args_to_floats(args)?;
+            match rule.evaluate(&floats) {
+                Ok(result) => Some(bool_term(result)),
+                Err(_) => None,
+            }
+        });
+    }
+
+    evaluator = evaluator.with_custom_function(NamedNode::new(FN_ABS_DIFF).unwrap(), |args| {
+        if args.len() != 2 {
+            return None;
+        }
+        let left = term_to_f64(&args[0])?;
+        let right = term_to_f64(&args[1])?;
+        Some(decimal_term(abs_diff(left, right)))
+    });
+
+    evaluator =
+        evaluator.with_custom_function(NamedNode::new(FN_RELATIVE_CHANGE).unwrap(), |args| {
+            if args.len() != 2 {
+                return None;
+            }
+            let left = term_to_f64(&args[0])?;
+            let right = term_to_f64(&args[1])?;
+            let result = relative_change(left, right);
+            if result.is_finite() {
+                Some(decimal_term(result))
+            } else {
+                None
+            }
+        });
+
+    evaluator = evaluator.with_custom_function(NamedNode::new(FN_ZSCORE).unwrap(), |args| {
+        if args.len() != 3 {
+            return None;
+        }
+        let value = term_to_f64(&args[0])?;
+        let mean = term_to_f64(&args[1])?;
+        let sigma = term_to_f64(&args[2])?;
+        Some(decimal_term(zscore(value, mean, sigma)))
+    });
+
+    evaluator
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use oxigraph::model::{GraphName, Quad};
+    use oxigraph::sparql::QueryResults;
+    use oxigraph::store::Store;
+
+    fn named(uri: &str) -> NamedNode {
+        NamedNode::new(uri).unwrap()
+    }
+
+    fn lit(value: &str) -> Term {
+        Term::Literal(Literal::new_simple_literal(value))
+    }
+
+    fn insert(store: &Store, subject: &str, predicate: &str, object: &str) {
+        store
+            .insert(&Quad::new(
+                named(subject),
+                named(predicate),
+                lit(object),
+                GraphName::DefaultGraph,
+            ))
+            .unwrap();
+    }
+
+    #[test]
+    fn absolute_threshold_filter_works_in_real_oxigraph_query() {
+        let store = Store::new().unwrap();
+        insert(&store, "https://janus.rs/test#sensor1", "https://janus.rs/test#live", "2.5");
+        insert(&store, "https://janus.rs/test#sensor1", "https://janus.rs/test#hist", "1.0");
+        insert(&store, "https://janus.rs/test#sensor2", "https://janus.rs/test#live", "1.1");
+        insert(&store, "https://janus.rs/test#sensor2", "https://janus.rs/test#hist", "1.0");
+
+        let query = r#"
+            PREFIX janus: <https://janus.rs/fn#>
+            PREFIX test:  <https://janus.rs/test#>
+            SELECT ?sensor WHERE {
+                ?sensor test:live ?live ;
+                        test:hist ?hist .
+                FILTER(janus:absolute_threshold_exceeded(?live, ?hist, 0.2))
+            }
+        "#;
+
+        let evaluator = build_evaluator();
+        let parsed_query = evaluator.parse_query(query).unwrap();
+        let results = parsed_query.on_store(&store).execute().unwrap();
+
+        if let QueryResults::Solutions(solutions) = results {
+            let rows: Vec<_> = solutions.collect::<Result<Vec<_>, _>>().unwrap();
+            assert_eq!(rows.len(), 1);
+            let sensor = rows[0].get("sensor").unwrap();
+            assert_eq!(sensor.to_string(), "<https://janus.rs/test#sensor1>");
+        } else {
+            panic!("expected SELECT solutions");
+        }
+    }
+}

src/extensions/registry.rs

@@ -0,0 +1,75 @@
+//! Registry of Janus extension-function names.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use crate::extensions::rules::{
+    AbsoluteThreshold, CatchUp, ExtensionRule, IsOutlier, RelativeThreshold, TrendDivergent,
+    VolatilityIncrease,
+};
+
+/// Namespace shared by all Janus extension functions.
+pub const JANUS_NS: &str = "https://janus.rs/fn#";
+
+/// Fully qualified URIs for registered scalar and boolean functions.
+pub const FN_ABS_DIFF: &str = "https://janus.rs/fn#abs_diff";
+pub const FN_RELATIVE_CHANGE: &str = "https://janus.rs/fn#relative_change";
+pub const FN_ZSCORE: &str = "https://janus.rs/fn#zscore";
+pub const FN_ABSOLUTE_THRESHOLD: &str = "https://janus.rs/fn#absolute_threshold_exceeded";
+pub const FN_RELATIVE_THRESHOLD: &str = "https://janus.rs/fn#relative_threshold_exceeded";
+pub const FN_CATCH_UP: &str = "https://janus.rs/fn#catch_up";
+pub const FN_VOLATILITY_INCREASE: &str = "https://janus.rs/fn#volatility_increase";
+pub const FN_IS_OUTLIER: &str = "https://janus.rs/fn#is_outlier";
+pub const FN_TREND_DIVERGENT: &str = "https://janus.rs/fn#trend_divergent";
+
+/// Registry that maps Janus function URIs to boolean rule implementations.
+pub struct FunctionRegistry {
+    rules: HashMap<&'static str, Arc<dyn ExtensionRule>>,
+}
+
+impl FunctionRegistry {
+    /// Build the default registry with all boolean rules pre-populated.
+    pub fn new() -> Self {
+        let mut rules: HashMap<&'static str, Arc<dyn ExtensionRule>> = HashMap::new();
+        rules.insert(FN_ABSOLUTE_THRESHOLD, Arc::new(AbsoluteThreshold));
+        rules.insert(FN_RELATIVE_THRESHOLD, Arc::new(RelativeThreshold));
+        rules.insert(FN_CATCH_UP, Arc::new(CatchUp));
+        rules.insert(FN_VOLATILITY_INCREASE, Arc::new(VolatilityIncrease));
+        rules.insert(FN_IS_OUTLIER, Arc::new(IsOutlier));
+        rules.insert(FN_TREND_DIVERGENT, Arc::new(TrendDivergent));
+        Self { rules }
+    }
+
+    /// Look up a boolean rule by its fully-qualified URI.
+    pub fn lookup(&self, name: &str) -> Option<&dyn ExtensionRule> {
+        self.rules.get(name).map(Arc::as_ref)
+    }
+
+    /// Iterate over all registry entries.
+    pub fn all_rules(&self) -> impl Iterator<Item = (&'static str, Arc<dyn ExtensionRule>)> + '_ {
+        self.rules.iter().map(|(key, value)| (*key, Arc::clone(value)))
+    }
+}
+
+impl Default for FunctionRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lookup_known_rule_returns_some() {
+        let registry = FunctionRegistry::new();
+        assert!(registry.lookup(FN_ABSOLUTE_THRESHOLD).is_some());
+    }
+
+    #[test]
+    fn lookup_unknown_name_returns_none() {
+        let registry = FunctionRegistry::new();
+        assert!(registry.lookup("https://janus.rs/fn#nonexistent").is_none());
+    }
+}