fuse-rule 0.1.0

High-performance, Arrow-native Complex Event Processing (CEP) engine with SQL-powered rules
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

use arrow::datatypes::{DataType, Field, Schema};
use fuse_rule::evaluator::{DataFusionEvaluator, RuleEvaluator};
use fuse_rule::rule::Rule;
use std::sync::Arc;

#[test]
fn test_predicate_compilation() {
    let evaluator = DataFusionEvaluator::new();
    let schema = Schema::new(vec![
        Field::new("price", DataType::Float64, true),
        Field::new("volume", DataType::Int32, true),
    ]);

    let rule = Rule {
        id: "test".to_string(),
        name: "Test".to_string(),
        predicate: "price > 100 AND volume < 50".to_string(),
        action: "logger".to_string(),
        window_seconds: None,
        version: 1,
        enabled: true,
    };

    let compiled = evaluator.compile(rule, &Arc::new(schema));
    assert!(compiled.is_ok());
}

#[test]
fn test_aggregate_detection() {
    let evaluator = DataFusionEvaluator::new();
    let schema = Schema::new(vec![Field::new("price", DataType::Float64, true)]);

    let rule_with_agg = Rule {
        id: "test_agg".to_string(),
        name: "Aggregate Test".to_string(),
        predicate: "AVG(price) > 100".to_string(),
        action: "logger".to_string(),
        window_seconds: Some(60),
        version: 1,
        enabled: true,
    };

    let compiled = evaluator.compile(rule_with_agg, &Arc::new(schema));
    assert!(compiled.is_ok());
    assert!(compiled.unwrap().has_aggregates);
}

#[test]
fn test_invalid_predicate() {
    let evaluator = DataFusionEvaluator::new();
    let schema = Schema::new(vec![Field::new("price", DataType::Float64, true)]);

    let rule = Rule {
        id: "test".to_string(),
        name: "Test".to_string(),
        predicate: "nonexistent_field > 100".to_string(),
        action: "logger".to_string(),
        window_seconds: None,
        version: 1,
        enabled: true,
    };

    let compiled = evaluator.compile(rule, &Arc::new(schema));
    assert!(compiled.is_err());
}