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

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

//! Validate B-tree key comparisons before publishing a candidate row.

use super::{index_key_values, index_predicate_accepts, PhysicalIndexDefinitions};
use crate::mutation::constraints::{
    index_keys::{changes_error, key_values_equal},
    ConstraintContext,
};
use std::collections::BTreeSet;
use uqa_core::{DocId, Predicate, Value};
use uqa_sql::{ast::IndexKey, SQLError};
use uqa_storage::{document_store::Document, ValueIndexKey};

impl PhysicalIndexDefinitions {
    pub fn validate_key_comparisons(
        &self,
        context: ConstraintContext<'_>,
        table: &str,
        document: &Document,
        ignored_doc_id: Option<DocId>,
    ) -> Result<(), SQLError> {
        for ((_, physical_key), index) in &self.indexes {
            if index.table != table
                || !index.method.eq_ignore_ascii_case("btree")
                || !index
                    .definition
                    .key_types
                    .iter()
                    .any(uqa_sql::expr::type_comparison_can_fail)
            {
                continue;
            }
            let predicate = index.definition.predicate.as_deref();
            let expressions = context.index_expressions();
            if !index_predicate_accepts(expressions, table, predicate, document)? {
                continue;
            }
            let values = index_key_values(expressions, table, &index.keys, document)?;
            let (key, probe) = if index.keys.iter().any(|key| key.column().is_none()) {
                (
                    ValueIndexKey::Index(physical_key.clone()),
                    Predicate::Equals(Value::Row(values.clone().into())),
                )
            } else {
                let Some(IndexKey::Column(first)) = index.keys.first() else {
                    continue;
                };
                let probe = match &values[0] {
                    Value::Null => Predicate::IsNull,
                    value => Predicate::Equals(value.clone()),
                };
                (ValueIndexKey::Column(first.clone()), probe)
            };
            // Schema validation compares an existing row only with other rows. An index probe would compare the row with its own entry before that identity could be excluded.
            let candidates: BTreeSet<DocId> = if ignored_doc_id.is_some() {
                context
                    .reads
                    .live_table_doc_ids(table)?
                    .into_iter()
                    .collect()
            } else {
                match context.indexes.value_index_scan_key(table, &key, &probe)? {
                    Some(list) => list.entries().iter().map(|entry| entry.doc_id).collect(),
                    None => context
                        .reads
                        .live_table_doc_ids(table)?
                        .into_iter()
                        .collect(),
                }
            };
            let changes = context
                .reads
                .command_overlay_changes(table)?
                .unwrap_or_default();
            let mut changed = changes.changes().peekable();
            let mut candidates = candidates.into_iter().peekable();
            // The candidates and the changed rows in identity order, each identity once.
            loop {
                let next_changed = match changed.peek() {
                    Some(Ok((id, _))) => Some(*id),
                    Some(Err(_)) => {
                        let Some(Err(error)) = changed.next() else {
                            unreachable!("a peeked failure");
                        };
                        return Err(changes_error(error));
                    }
                    None => None,
                };
                let id = match (candidates.peek().copied(), next_changed) {
                    (None, None) => break,
                    (Some(candidate), Some(changed_id)) if changed_id < candidate => {
                        changed.next();
                        changed_id
                    }
                    (Some(candidate), next_changed) => {
                        candidates.next();
                        if next_changed == Some(candidate) {
                            changed.next();
                        }
                        candidate
                    }
                    (None, Some(changed_id)) => {
                        changed.next();
                        changed_id
                    }
                };
                if ignored_doc_id == Some(id) {
                    continue;
                }
                let Some(stored) = context.reads.get_document(table, id)? else {
                    continue;
                };
                if index_predicate_accepts(expressions, table, predicate, &stored)? {
                    key_values_equal(
                        &index_key_values(expressions, table, &index.keys, &stored)?,
                        &values,
                    )?;
                }
            }
        }
        Ok(())
    }
}