hawdb-executor 0.27.0

An embedded Rust graph database for local data-plane workloads
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

// Copyright 2026 Nowledge
// SPDX-License-Identifier: Apache-2.0

use super::*;
use hawdb_core::RuntimeCancellationToken;
use hawdb_storage::{RelId, RelRecord};

#[test]
fn relationship_count_source_releases_admission_on_every_exit() {
    for mode in [
        "complete",
        "stop",
        "error",
        "cancel",
        "mid_cancel",
        "refuse",
    ] {
        let cancellation = RuntimeCancellationToken::new();
        let task = RuntimeTaskContext::without_deadline(cancellation.clone());
        let mut catalog = Catalog::default();
        let label = catalog.get_or_create_label("Entity");
        let rel_type = catalog.get_or_create_rel_type("RELATES_TO");
        let key = Value::String("k".repeat(8192));
        let store = store::ReadFixture {
            nodes: (0..2)
                .map(|id| NodeRecord {
                    id: NodeId(id),
                    labels: [label].into_iter().collect(),
                    properties: BTreeMap::from([
                        ("id".into(), key.clone()),
                        ("body".into(), Value::String("x".repeat(32_768))),
                    ]),
                })
                .collect(),
            relationships: vec![RelRecord {
                id: RelId(0),
                source: NodeId(0),
                target: NodeId(1),
                rel_type,
                properties: BTreeMap::new(),
            }],
            adjacency_cancellation: (mode == "mid_cancel").then(|| cancellation.clone()),
            ..store::ReadFixture::default()
        };
        let plan = PhysicalPlan::OptionalRelationshipCountSumExec {
            variable: "e".into(),
            label: "Entity".into(),
            properties: BTreeMap::from([("id".into(), key)]),
            legs: vec![RelationshipCountLeg {
                rel_type: "RELATES_TO".into(),
                direction: RelationshipDirection::Outgoing,
                distinct: true,
                filter: None,
            }],
            output: "count".into(),
        };
        let memory = ExecutionMemoryConfig {
            query_memory_bytes: NonZeroUsize::new(if mode == "refuse" { 4096 } else { 65_536 })
                .unwrap(),
            batch_payload_bytes: NonZeroUsize::new(4096).unwrap(),
            batch_rows: NonZeroUsize::new(1).unwrap(),
            ..ExecutionMemoryConfig::default()
        };
        let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
        if mode == "cancel" {
            cancellation.cancel();
        }
        let mut emitted = 0;
        let result = with_context(None, |context| {
            dispatch_batch_operator(
                &plan,
                BatchExecution {
                    context: BatchReadContext {
                        catalog: &catalog,
                        store: &store,
                        memory: &memory,
                        memory_ledger: &ledger,
                        task_context: Some(&task),
                        ..context
                    },
                    execution_limit: ExecutionLimit::unlimited(),
                    emit: &mut |batch| {
                        emitted += batch.len();
                        assert_eq!(batch[0].values["count"], Value::Int(1));
                        match mode {
                            "stop" => Ok(BatchControl::Stop),
                            "error" => Err(HawDBError::Execution("consumer refused".into())),
                            _ => Ok(BatchControl::Continue),
                        }
                    },
                },
            )
        });
        match mode {
            "complete" => {
                assert_eq!(result.unwrap(), BatchControl::Continue);
                assert_eq!(emitted, 1);
            }
            "stop" => {
                assert_eq!(result.unwrap(), BatchControl::Stop);
                assert_eq!(emitted, 1);
            }
            "error" => {
                assert!(result.is_err());
                assert_eq!(emitted, 1);
            }
            "cancel" | "mid_cancel" => {
                assert!(
                    result.unwrap_err().to_string().contains("cancelled"),
                    "{mode}"
                );
                assert_eq!(emitted, 0);
            }
            "refuse" => {
                let Err(HawDBError::ResourceBudgetExceeded(error)) = result else {
                    panic!("expected source refusal, got {result:?}");
                };
                assert_eq!(error.owner, "OptionalRelationshipCountSumExec");
                assert_eq!(emitted, 0);
            }
            _ => unreachable!(),
        }
        assert_eq!(ledger.snapshot().used_bytes, 0, "{mode}");
    }
}