uqa-execution 0.4.8

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

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

//! Copy projected rows while an ordered provider cursor holds its storage borrow.

use super::{ScoredDocumentSource, ScoredInputCursor};
use crate::{ExecResult, PhysicalRow, ProjectedRow};
use uqa_sql::SQLError;

impl ScoredDocumentSource {
    pub(super) fn next_borrowed_physical_batch(
        &mut self,
        max_rows: usize,
    ) -> ExecResult<Option<Vec<PhysicalRow>>> {
        if self.lock_origin.is_some()
            || self.recheck_pinned
            || crate::query::generated::projection_contains_virtual_generated_column(
                &self.column_definitions,
                &self.projected_fields,
            )
            || crate::query::document_projection::projections_use_tuple_xmin(
                &self.projected_fields,
                &self.column_definitions,
            )
        {
            return Ok(None);
        }
        let ScoredInputCursor::All { mut after } = self.input else {
            return Ok(None);
        };
        if max_rows == 0 {
            return Ok(Some(Vec::new()));
        }
        self.serializable.observe_relation()?;
        let fields = self
            .projected_fields
            .iter()
            .map(String::as_str)
            .collect::<Vec<_>>();
        let mut rows = Vec::with_capacity(max_rows);
        while rows.len() < max_rows {
            let remaining = max_rows - rows.len();
            let mut last = None;
            let mut failure = None;
            let visited = self
                .table
                .read_documents()
                .for_each_next_fields_borrowed(after, remaining, &fields, &mut |id, values| {
                    last = Some(id);
                    // Only positional evaluation and owned row construction are allowed here: neither can reenter the provider.
                    let result = (|| -> ExecResult<()> {
                        if let Some(predicate) = &self.predicate {
                            if !predicate.keep(values)? {
                                return Ok(());
                            }
                        }
                        let metadata = self.row_metadata(id, 0.0);
                        let extras = [metadata.doc_id()?, metadata.score(), metadata.table_oid()];
                        let row =
                            ProjectedRow::new(&self.schema, &self.projected_slots, values, &extras);
                        let row = PhysicalRow::from_values(row.into_values());
                        rows.push(self.append_metadata_attributes(row, id, 0.0)?);
                        Ok(())
                    })();
                    if let Err(error) = result {
                        failure = Some(error);
                        return false;
                    }
                    true
                })
                .map_err(|error| {
                    SQLError::Internal(format!(
                        "scan `{}` borrowed projected documents: {error}",
                        self.table_name
                    ))
                });
            if let Some(error) = failure {
                return Err(error);
            }
            let Some(visited) = visited? else {
                return Ok((!rows.is_empty()).then_some(rows));
            };
            if let Some(last) = last {
                after = Some(last);
                self.input = ScoredInputCursor::All { after };
            }
            if visited == 0 {
                break;
            }
        }
        Ok(Some(rows))
    }
}