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

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

//! Execute compiled statements with planning and statement scopes captured at their point of use.

use super::{context::StatementExecutionInputs, plan_executor::UnifiedPlanExecutor};
use uqa_sql::catalog::roles::RoleReference;
use uqa_sql::{
    binding::statements::AnalyzedResult,
    binding::stored_routines::mark_catalog_statement_relations_bound,
    plan::{AggregateClassifier, CommandPlan, ExecutablePlanOptimizer, UnifiedPlan},
    SQLError, SQLParam, SQLResult, Statement,
};

pub struct CompiledStatementContext<'a, S: Clone + 'static> {
    pub aggregates: &'a dyn AggregateClassifier,
    pub planning: &'a dyn ExecutablePlanOptimizer,
    pub statements: &'a dyn StatementExecutionInputs<S>,
}

pub fn execute<S: Clone + Send + Sync + 'static>(
    context: &CompiledStatementContext<'_, S>,
    statement: Statement,
    params: &[SQLParam],
) -> Result<SQLResult, SQLError> {
    let plan = UnifiedPlan::lower_with(statement, context.aggregates);
    execute_plan(context, plan, params)
}

pub fn execute_plan<S: Clone + Send + Sync + 'static>(
    context: &CompiledStatementContext<'_, S>,
    plan: UnifiedPlan,
    params: &[SQLParam],
) -> Result<SQLResult, SQLError> {
    let _temporal_input = super::context::temporal_input_scope(context.statements);
    let plan = context.planning.plan_for_execution(plan, params)?;
    UnifiedPlanExecutor::new_nested(context.statements.statement_execution_context(), params)
        .execute(&plan)
}

/// Run a statement once `check` accepts what its analysis derives about its result.
pub fn execute_checked_plan<S: Clone + Send + Sync + 'static>(
    context: &CompiledStatementContext<'_, S>,
    plan: UnifiedPlan,
    params: &[SQLParam],
    check: &dyn Fn(&AnalyzedResult) -> Result<(), SQLError>,
) -> Result<SQLResult, SQLError> {
    let _temporal_input = super::context::temporal_input_scope(context.statements);
    let (plan, result) = context.planning.plan_with_result(plan, params)?;
    check(&result)?;
    UnifiedPlanExecutor::new_nested(context.statements.statement_execution_context(), params)
        .execute(&plan)
}

pub fn execute_with_privilege_subject<S: Clone + Send + Sync + 'static>(
    context: &CompiledStatementContext<'_, S>,
    statement: Statement,
    params: &[SQLParam],
    privilege_subject: &RoleReference,
) -> Result<SQLResult, SQLError> {
    let _temporal_input = super::context::temporal_input_scope(context.statements);
    let mut plan = UnifiedPlan::lower_with(statement, context.aggregates);
    mark_catalog_statement_relations_bound(&mut plan)?;
    let plan = context.planning.plan_for_execution(plan, params)?;
    UnifiedPlanExecutor::new_nested(context.statements.statement_execution_context(), params)
        .with_privilege_subject(privilege_subject)
        .execute(&plan)
}

pub fn execute_optimized_command<S: Clone + Send + Sync + 'static>(
    statements: &dyn StatementExecutionInputs<S>,
    command: &CommandPlan,
    params: &[SQLParam],
) -> Result<SQLResult, SQLError> {
    let _temporal_input = super::context::temporal_input_scope(statements);
    UnifiedPlanExecutor::new_nested(statements.statement_execution_context(), params)
        .execute(&UnifiedPlan::Command(Box::new(command.clone())))
}

#[cfg(test)]
mod tests;