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

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

//! First-use SQL statement analysis, separate from source parsing and executable planning.

use parking_lot::Mutex;
use std::{collections::BTreeMap, sync::Arc};
use uqa_sql::{
    prepared::definition::{analysis_is_current, PreparedDefinition, PreparedDefinitionContext},
    routines::SQLUserFunction,
    ColumnType, SQLError,
};

#[derive(Clone, Copy)]
pub struct SQLBodyIdentity {
    object: [u8; 16],
    version: u64,
}

impl SQLBodyIdentity {
    pub fn new(function: &SQLUserFunction) -> Result<Self, SQLError> {
        Ok(Self {
            object: function
                .def
                .object_id
                .ok_or_else(|| SQLError::Internal("SQL body has no routine identity".into()))?,
            version: function.definition_version()?,
        })
    }
}

struct StatementInputs {
    position: usize,
    parameter_types: Vec<ColumnType>,
    definition: Arc<PreparedDefinition>,
}

struct RoutineInputs {
    version: u64,
    statements: Vec<StatementInputs>,
}

/// Session-local analysis by routine revision, statement position and concrete input types.
#[derive(Default)]
pub struct SQLRoutineInputs {
    routines: Mutex<BTreeMap<[u8; 16], RoutineInputs>>,
}

pub struct SQLRoutineInputContext<'a> {
    pub cache: &'a SQLRoutineInputs,
    pub analysis: PreparedDefinitionContext<'a>,
}

impl SQLRoutineInputs {
    pub(crate) fn invalidate(
        &self,
        affects: impl Fn(&uqa_sql::prepared::dependencies::PreparedAnalysisDependencies) -> bool,
    ) {
        let mut routines = self.routines.lock();
        for routine in routines.values_mut() {
            routine
                .statements
                .retain(|statement| !affects(&statement.definition.dependencies));
        }
        routines.retain(|_, routine| !routine.statements.is_empty());
    }

    /// Publish only successful analysis and result validation, before execution.
    /// Neither catalog callbacks nor recursive preparation run under the cache lock.
    pub fn statement(
        &self,
        identity: SQLBodyIdentity,
        position: usize,
        parameter_types: &[ColumnType],
        context: &PreparedDefinitionContext<'_>,
        prepare: impl FnOnce() -> Result<PreparedDefinition, SQLError>,
    ) -> Result<Arc<PreparedDefinition>, SQLError> {
        let retained = self
            .routines
            .lock()
            .get(&identity.object)
            .filter(|routine| routine.version == identity.version)
            .and_then(|routine| {
                routine.statements.iter().find(|statement| {
                    statement.position == position && statement.parameter_types == parameter_types
                })
            })
            .map(|statement| Arc::clone(&statement.definition));
        if let Some(definition) = retained {
            if analysis_is_current(
                context,
                definition.effective_search_path.as_ref(),
                &definition.dependencies,
                definition.dependency_snapshot.as_ref(),
            )? {
                return Ok(definition);
            }
        }
        let definition = Arc::new(prepare()?);
        let mut routines = self.routines.lock();
        let routine = routines
            .entry(identity.object)
            .or_insert_with(|| RoutineInputs {
                version: identity.version,
                statements: Vec::new(),
            });
        if routine.version != identity.version {
            routine.version = identity.version;
            routine.statements.clear();
        }
        routine.statements.retain(|statement| {
            statement.position != position || statement.parameter_types != parameter_types
        });
        routine.statements.push(StatementInputs {
            position,
            parameter_types: parameter_types.to_vec(),
            definition: Arc::clone(&definition),
        });
        Ok(definition)
    }
}

#[cfg(test)]
mod tests;