uqa-execution 0.5.0

Volcano physical operators with row-batch pipelines
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Reconcile selected storage indexes with cached command keys and retained private versions.

use super::document_changes::DocumentChanges;
use crate::mutation::overlay::CommandIndexProbe;
use crate::storage_errors::storage_error;
use uqa_core::{CancellationToken, Payload, PostingEntry, PostingList, Predicate, Value};
use uqa_sql::SQLError;
use uqa_storage::DocumentStore;

/// All inputs refer to the query's selected relation generation and original read allowance.
pub trait QueryIndexRead: Sync {
    /// Probe stored values and record the original serializable predicate observation.
    fn value_index_scan(
        &self,
        table: &str,
        field: &str,
        predicate: &Predicate,
    ) -> Result<Option<PostingList>, SQLError>;
    fn command_overlay_changes(&self, table: &str) -> Result<Option<DocumentChanges>, SQLError>;
    /// Capture command-only masks together with cached matches from the same frame stack.
    fn exact_command_matches(
        &self,
        table: &str,
        field: &str,
        value: &Value,
    ) -> Result<CommandIndexProbe, SQLError>;
}

pub fn scan(
    reads: &dyn QueryIndexRead,
    table: &str,
    field: &str,
    predicate: &Predicate,
    cancellation: &CancellationToken,
) -> Result<Option<PostingList>, SQLError> {
    let Some(indexed) = reads.value_index_scan(table, field, predicate)? else {
        return Ok(None);
    };
    let changes = reads.command_overlay_changes(table)?;
    if changes.as_ref().is_some_and(DocumentChanges::has_changes) {
        if let Predicate::Equals(value) = predicate {
            if crate::catalog::index::value::field_is_index_safe(value) {
                if matches!(value, Value::Null) {
                    return Ok(Some(PostingList::from_sorted_unchecked(Vec::new())));
                }
                let command = reads.exact_command_matches(table, field, value)?;
                return merge_exact(
                    indexed,
                    &changes.expect("changed rows").without_staged(),
                    command,
                    field,
                    predicate,
                    cancellation,
                )
                .map(Some);
            }
        }
    }
    merge_changes(indexed, changes, field, predicate, cancellation).map(Some)
}

fn merge_exact(
    indexed: PostingList,
    fixed: &DocumentChanges,
    command: CommandIndexProbe,
    field: &str,
    predicate: &Predicate,
    cancellation: &CancellationToken,
) -> Result<PostingList, SQLError> {
    let error = |error| storage_error("read indexed query changes", &error);
    let mut entries = Vec::with_capacity(indexed.len());
    for entry in indexed.entries() {
        cancellation.check()?;
        if !command
            .changes
            .contains_change(entry.doc_id)
            .map_err(error)?
            && !fixed.contains_change(entry.doc_id).map_err(error)?
        {
            entries.push(entry.clone());
        }
    }
    // Fixed portal/transaction versions retain their original source. Command keys above them are never obtained by traversing these changes.
    for change in fixed.changes() {
        cancellation.check()?;
        let (id, present) = change.map_err(error)?;
        if present
            && !command.changes.contains_change(id).map_err(error)?
            && predicate.evaluate(fixed.get_field(id, field).map_err(error)?.as_ref())
        {
            entries.push(PostingEntry::new(id, Payload::default()));
        }
    }
    for &id in command.matches.iter() {
        cancellation.check()?;
        entries.push(PostingEntry::new(id, Payload::default()));
    }
    entries.sort_unstable_by_key(|entry| entry.doc_id);
    Ok(PostingList::from_sorted_unchecked(entries))
}

fn merge_changes(
    indexed: PostingList,
    changes: Option<DocumentChanges>,
    field: &str,
    predicate: &Predicate,
    cancellation: &CancellationToken,
) -> Result<PostingList, SQLError> {
    let Some(changes) = changes.filter(DocumentChanges::has_changes) else {
        return Ok(indexed);
    };
    let error = |error| storage_error("read indexed command changes", &error);
    let mut candidates = indexed.entries().iter().peekable();
    let mut entries = Vec::with_capacity(indexed.len());
    for change in changes.changes() {
        cancellation.check()?;
        let (id, present) = change.map_err(error)?;
        while candidates.peek().is_some_and(|entry| entry.doc_id < id) {
            entries.push(candidates.next().expect("a peeked candidate").clone());
        }
        if candidates.peek().is_some_and(|entry| entry.doc_id == id) {
            candidates.next();
        }
        if present && predicate.evaluate(changes.get_field(id, field).map_err(error)?.as_ref()) {
            entries.push(PostingEntry::new(id, Payload::default()));
        }
    }
    entries.extend(candidates.cloned());
    Ok(PostingList::from_sorted_unchecked(entries))
}

#[cfg(test)]
mod tests;