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

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

//! Stored values follow a composite type's attribute changes: every table column and materialized view column whose declared type holds values of the type is rewritten.

use std::collections::BTreeMap;

use crate::mutation::constraints::context::MutationRead;
use crate::schema::columns::ColumnRewritePublication;
use crate::schema::view_dependencies::ViewDependencyContext;
use uqa_sql::ast::ColumnDef;
use uqa_sql::expr::composites::{
    apply_attribute_change, type_contains_composite, AttributeChange, CompositeTypeCatalog,
};
use uqa_sql::SQLError;

/// The tables of the catalog with their declared columns.
pub trait CompositeValueTables {
    fn composite_value_tables(&self) -> Result<Vec<(String, Vec<ColumnDef>)>, SQLError>;
}

pub struct CompositeValueContext<'a> {
    pub memory: &'a dyn crate::query::runtime::QueryMemorySettings,
    pub cancellation: &'a uqa_core::CancellationToken,
    pub tables: &'a dyn CompositeValueTables,
    pub reads: &'a dyn MutationRead,
    pub writes: &'a dyn ColumnRewritePublication,
    pub views: ViewDependencyContext<'a>,
    /// The attributes of every composite type before the change.
    pub types: &'a dyn CompositeTypeCatalog,
    pub catalogs: super::catalog_values::CompositeCatalogValueContext<'a>,
}

/// Rewrite the stored values of composite type `target` for `change`, in tables and materialized views.
pub fn rewrite_composite_values(
    context: &CompositeValueContext<'_>,
    target: u32,
    change: &AttributeChange,
) -> Result<std::collections::BTreeSet<String>, SQLError> {
    let memory = uqa_core::memory::MemoryBudget::new(context.memory.work_mem_bytes()?);
    let control = uqa_storage::read_control::StorageReadControl::new(&memory, context.cancellation);
    let mut rebuild = std::collections::BTreeSet::new();
    for (table, columns) in context.tables.composite_value_tables()? {
        let mut affected = Vec::new();
        for column in columns {
            if type_contains_composite(&column.ty, target, context.types)? {
                affected.push(column);
            }
        }
        if affected.is_empty() {
            continue;
        }
        rebuild.insert(table.clone());
        crate::schema::columns::rows::visit_document_ids(
            context.reads,
            &table,
            &control,
            &mut |doc_id| {
                let Some(document) = context.reads.get_document(&table, doc_id)? else {
                    return Ok(());
                };
                let mut updates = BTreeMap::new();
                for column in &affected {
                    let Some(value) = document.get(&column.name).cloned() else {
                        continue;
                    };
                    updates.insert(
                        column.name.clone(),
                        apply_attribute_change(value, &column.ty, target, change, context.types)?,
                    );
                }
                if !updates.is_empty() {
                    context
                        .writes
                        .update_fields(&table, doc_id, updates, BTreeMap::new())?;
                }
                Ok(())
            },
        )?;
    }
    let constants = uqa_sql::expr::composites::constants::CompositeConstantChange {
        target,
        change,
        catalog: context.types,
        types: context.catalogs.types,
    };
    rebuild.extend(super::catalog_values::rewrite(
        &context.catalogs,
        &constants,
    )?);
    crate::schema::view_dependencies::rewrite_materialized_composite_values(
        &context.views,
        target,
        change,
        context.types,
        &constants,
    )?;
    Ok(rebuild)
}

pub(super) fn rebuild_indexes(
    context: &CompositeValueContext<'_>,
    tables: std::collections::BTreeSet<String>,
) -> Result<(), SQLError> {
    for table in tables {
        context
            .catalogs
            .index_publication
            .refresh_index_table(&table)
            .map_err(|error| {
                uqa_sql::catalog::errors::storage_error("rebuild composite index values", &error)
            })?;
    }
    Ok(())
}