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

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

//! Scalar access candidates preserve the whole residual predicate and source column mapping.

use super::indexed_filter::{self, QueryIndexRead};
use uqa_core::{CancellationToken, DocId};
use uqa_sql::{
    ast::ColumnDef,
    retrieval::{RetrievalConstants, RetrievalExpr},
    SQLError, SQLParam, ScalarExpr,
};

pub(crate) struct IndexCandidates<'a> {
    pub reads: &'a dyn QueryIndexRead,
    pub table: &'a str,
    pub columns: &'a [ColumnDef],
    /// The source's visible name for a stored column, including a column alias.
    pub visible: &'a dyn Fn(&str) -> String,
    pub params: &'a [SQLParam],
    pub command_visible: bool,
    pub cancellation: &'a CancellationToken,
}

impl IndexCandidates<'_> {
    pub(crate) fn select(&self, filter: &ScalarExpr) -> Result<Option<Vec<DocId>>, SQLError> {
        self.cancellation.check()?;
        if let ScalarExpr::And(parts) = filter {
            // One answerable conjunct bounds the scan. The whole filter remains a residual, including other conjuncts and comparisons that can fail.
            for part in parts {
                if let Some(ids) = self.select(part)? {
                    return Ok(Some(ids));
                }
            }
            return Ok(None);
        }
        let lowered = uqa_sql::retrieval::lower_where(
            filter,
            &RetrievalConstants {
                params: self.params,
                evaluate: &|expression, params| {
                    if !uqa_sql::semantics::mutation_inputs::expr_is_row_independent(expression) {
                        return Err(SQLError::Internal("nonconstant index bound".into()));
                    }
                    crate::eval_scalar(expression, &crate::ScalarEvalContext::new(None, params))
                },
                stores: &|name| {
                    self.columns
                        .iter()
                        .any(|column| (self.visible)(&column.name) == name)
                },
            },
        );
        let Some(RetrievalExpr::Filter {
            field,
            predicate,
            source: None,
        }) = lowered
        else {
            return Ok(None);
        };
        let Some(column) = self
            .columns
            .iter()
            .find(|column| (self.visible)(&column.name) == field)
        else {
            return Ok(None);
        };
        let indexed = if self.command_visible {
            indexed_filter::scan(
                self.reads,
                self.table,
                &column.name,
                &predicate,
                self.cancellation,
            )
        } else {
            self.reads
                .value_index_scan(self.table, &column.name, &predicate)
        };
        indexed.map(|result| {
            result.map(|posting| posting.entries().iter().map(|entry| entry.doc_id).collect())
        })
    }
}

#[cfg(test)]
mod tests;