uqa-execution 0.3.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

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

//! Select changed routine definitions and publish their freshly compiled bodies in catalog order.

use super::{
    catalog::{RoutineRegistryPublication, RoutineRegistryState},
    compilation::{compile_persisted_sql_function, StoredRoutineCompilationContext},
};
use crate::schema::namespaces::NamespaceCatalogChanges;
use std::sync::Arc;
use uqa_core::RelationIdentity;
use uqa_sql::{
    ast::CreateFunction,
    routines::{
        lifecycle::rewrites::{self as analysis},
        merge_columns::statement_has_removed_merge_target,
        SQLUserFunction,
    },
    SQLError,
};

pub struct RoutineRewriteContext<'a> {
    pub registry: &'a dyn RoutineRegistryState,
    pub publication: &'a dyn RoutineRegistryPublication,
    pub compilation: StoredRoutineCompilationContext<'a>,
    pub columns: uqa_sql::binding::stored_columns::StoredColumnBindingContext<'a>,
    pub changes: &'a dyn NamespaceCatalogChanges,
}

pub fn rewrite_routine_relation_references(
    context: &RoutineRewriteContext<'_>,
    from: &RelationIdentity,
    to: &RelationIdentity,
) -> Result<(), SQLError> {
    rewrite_stored_routine_bodies(context, |statement| {
        uqa_sql::catalog::events::renames::rewrite_stored_statement_relation(statement, from, to)
    })
}

pub fn rewrite_routine_column_references(
    context: &RoutineRewriteContext<'_>,
    relation: &RelationIdentity,
    from: &str,
    to: &str,
) -> Result<(), SQLError> {
    rewrite_stored_routine_bodies(context, |statement| {
        uqa_sql::binding::stored_columns::rewrite_stored_statement_column(
            context.columns,
            statement,
            relation,
            from,
            to,
        )
    })
}

pub fn rewrite_stored_routine_bodies(
    context: &RoutineRewriteContext<'_>,
    rewrite: impl FnMut(&mut uqa_sql::ast::Statement) -> Result<bool, SQLError>,
) -> Result<(), SQLError> {
    let registry = context.registry.routine_snapshot();
    let definitions = analysis::rewritten_routine_definitions(&registry, rewrite)?;
    publish_stored_routine_body_rewrites(context, definitions)
}

pub fn publish_stored_routine_body_rewrites(
    context: &RoutineRewriteContext<'_>,
    definitions: Vec<CreateFunction>,
) -> Result<(), SQLError> {
    if definitions.is_empty() {
        return Ok(());
    }
    let mut rewritten = context.registry.routine_snapshot();
    for definition in definitions {
        let function = analysis::routine_body_rewrite_target(&mut rewritten, &definition)?;
        let compiled = compile_persisted_sql_function(&context.compilation, &definition)?;
        *function = Arc::new(SQLUserFunction {
            def: definition,
            compiled,
        });
    }
    context
        .publication
        .persist_routine_definitions(&rewritten)?;
    **context.registry.routines_write() = rewritten;
    context.changes.catalog_registry_changed();
    Ok(())
}

pub fn refresh_stored_merge_target_plans(
    context: &RoutineRewriteContext<'_>,
) -> Result<(), SQLError> {
    rewrite_stored_routine_bodies(context, |statement| {
        statement_has_removed_merge_target(context.compilation.analysis.merge, statement)
    })
}