uqa-execution 0.4.6

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

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

//! Owned projections hold admitted output through batch handoff or the consuming callback.

use super::{
    BTreeMap, BudgetedVec, DocId, MemoryReservation, RetainedDocuments, StorageBackendError,
    StorageBackendResult, StoredDocument, Value,
};

impl RetainedDocuments {
    pub(super) fn read_owned_row(
        &self,
        id: DocId,
        memory: &mut MemoryReservation,
    ) -> StorageBackendResult<Option<StoredDocument>> {
        self.0.control.check()?;
        let private = self.0.changes.contains_change(id);
        let source: &dyn uqa_storage::DocumentStore = if private {
            &self.0.changes
        } else {
            self.0.source.as_ref()
        };
        let mut page =
            uqa_storage::document_store::read_stored_documents(source, &[id], &self.0.control)?;
        let row = page.pop().expect("one validated whole-row slot");
        drop(page);
        let Some(row) = row else {
            return Ok(None);
        };
        let (row, allocation) = row.into_budgeted(&self.0.control)?.into_parts();
        memory.absorb(allocation);
        let row = if private {
            self.0.private_layout.complete_private(row, memory)
        } else {
            self.0.layout.adapt_base(row, memory)
        }
        .map_err(super::layout_error)?;
        self.0.control.check()?;
        Ok(Some(row))
    }

    fn copy_values(
        &self,
        values: &[&Value],
        memory: &mut MemoryReservation,
    ) -> StorageBackendResult<Vec<Value>> {
        self.0.control.check()?;
        let mut copied = BudgetedVec::new(self.0.control.memory());
        copied.reserve(values.len())?;
        for value in values {
            let (value, allocation) = value
                .clone_budgeted(self.0.control.memory(), self.0.control.cancellation())
                .map_err(|error| match error {
                    uqa_core::ValueRetentionError::Memory(error) => {
                        StorageBackendError::Memory(error)
                    }
                    uqa_core::ValueRetentionError::Cancelled(error) => {
                        StorageBackendError::Cancelled(error)
                    }
                })?
                .into_parts();
            memory.absorb(allocation);
            copied.push(value)?;
        }
        self.0.control.check()?;
        let (copied, allocation) = copied.into_parts();
        memory.absorb(allocation);
        Ok(copied)
    }

    pub(super) fn copy_projected_rows(
        &self,
        ids: &[DocId],
        fields: &[&str],
    ) -> StorageBackendResult<BTreeMap<DocId, Vec<Value>>> {
        let mut memory = self.0.control.memory().empty_reservation();
        let mut rows = BTreeMap::new();
        let mut failure = None;
        let read = self.visit_projection(ids, fields, &mut |id, present, values| {
            // A fixed view gives every duplicate identity the same values; the owned map needs only one copy.
            if !present || rows.contains_key(&id) {
                return true;
            }
            let copied = memory
                .grow(size_of::<(DocId, Vec<Value>)>())
                .map_err(StorageBackendError::from)
                .and_then(|()| self.copy_values(values, &mut memory));
            match copied {
                Ok(values) => {
                    rows.insert(id, values);
                    true
                }
                Err(error) => {
                    failure = Some(error);
                    false
                }
            }
        });
        if let Some(error) = failure {
            return Err(error);
        }
        read?;
        self.0.control.check()?;
        Ok(rows)
    }

    pub(super) fn visit_owned_projection(
        &self,
        ids: &[DocId],
        fields: &[&str],
        visitor: &mut dyn FnMut(DocId, Vec<Value>) -> bool,
    ) -> StorageBackendResult<()> {
        let mut failure = None;
        let read = self.visit_projection(ids, fields, &mut |id, _, values| {
            let mut memory = self.0.control.memory().empty_reservation();
            match self.copy_values(values, &mut memory) {
                Ok(values) => visitor(id, values),
                Err(error) => {
                    failure = Some(error);
                    false
                }
            }
        });
        if let Some(error) = failure {
            return Err(error);
        }
        read
    }
}