hawdb-executor 0.26.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

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

//! Expand a list for each upstream row without materializing the input relation.

use super::*;
use crate::expression::evaluate_projection_borrowed;
use crate::pipeline::AccountedBindingBatch;
use hawdb_core::{HawDBError, Value};

pub fn stream_unwind_rows_batches(
    expression: &ProjectionExpression,
    variable: &str,
    input: &PhysicalPlan,
    source: &mut dyn BindingBatchSource,
    context: BatchExecutionContext<'_>,
    execution_limit: ExecutionLimit,
    emit: &mut dyn FnMut(BindingBatch) -> Result<BatchControl>,
) -> Result<BatchControl> {
    runtime_checkpoint(context.task_context)?;
    if execution_limit.is_reached(0) {
        return Ok(BatchControl::Stop);
    }
    let working = context.source_account("UnwindRowsExec expressions");
    let mut output = AccountedBindingBatch::with_ledger(
        "UnwindRowsExec",
        context.memory.batch_rows.get(),
        context.memory.batch_payload_bytes,
        context.memory_ledger,
    );
    let mut produced = 0usize;
    let control = source.execute(input, ExecutionLimit::unlimited(), &mut |batch| {
        for binding in batch {
            runtime_checkpoint(context.task_context)?;
            if execution_limit.is_reached(produced) {
                return Ok(BatchControl::Stop);
            }
            // WITH may leave physically retained graph slots after their names
            // leave logical scope. A new scalar column can reuse such a name.
            if binding.values.contains_key(variable) {
                return Err(HawDBError::Semantic(format!(
                    "UNWIND variable '{variable}' is already bound"
                )));
            }
            let value = evaluate_projection_borrowed(
                expression,
                context.catalog,
                &binding,
                Some(&working),
            )?;
            let values = match value.as_ref() {
                Value::List(values) => values,
                Value::Null => continue,
                _ => {
                    return Err(HawDBError::Execution(
                        "UNWIND source must resolve to a list value or NULL".into(),
                    ))
                }
            };
            for value in values {
                runtime_checkpoint(context.task_context)?;
                if execution_limit.is_reached(produced) {
                    return Ok(BatchControl::Stop);
                }
                let bytes = binding_memory_bytes_with_values(
                    &binding,
                    binding
                        .values
                        .iter()
                        .map(|(name, value)| (name.as_str(), value))
                        .chain(std::iter::once((variable, value))),
                );
                if output.push_generated(
                    bytes,
                    || {
                        let mut row = binding.clone();
                        row.values.insert(variable.to_owned(), value.clone());
                        row
                    },
                    emit,
                )? == BatchControl::Stop
                {
                    return Ok(BatchControl::Stop);
                }
                produced = produced.saturating_add(1);
                if output.is_full() && output.emit(emit)? == BatchControl::Stop {
                    return Ok(BatchControl::Stop);
                }
            }
        }
        Ok(BatchControl::Continue)
    })?;
    if output.emit(emit)? == BatchControl::Stop {
        return Ok(BatchControl::Stop);
    }
    Ok(control)
}