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

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

//! Prepared registration ordered around retained session registry publication.

pub mod invalidation;

use std::sync::Arc;
use uqa_sql::{
    plan::{AggregateClassifier, UnifiedPlan},
    prepared::{definition::PreparedDefinitionContext, entry::PreparedStatementPlan},
    ColumnType, SQLError, Statement,
};

pub trait PreparedDefinitionWrite {
    fn insert(&mut self, name: String, definition: PreparedStatementPlan);
}

pub trait PreparedDefinitionRegistry {
    fn write_definitions(&self) -> Box<dyn PreparedDefinitionWrite + '_>;
}

#[derive(Clone, Copy)]
pub struct PreparedRegistrationContext<'a> {
    pub analysis: PreparedDefinitionContext<'a>,
    pub aggregates: &'a dyn AggregateClassifier,
    pub registry: &'a dyn PreparedDefinitionRegistry,
    /// The session's active transaction clock, including direct host registration.
    pub transaction_timestamp_micros: Option<i64>,
    pub date_order: Option<uqa_core::TemporalDateOrder>,
    pub clock: fn() -> i64,
}

pub fn register_statement(
    context: &PreparedRegistrationContext<'_>,
    name: String,
    statement: Statement,
) -> Result<(), SQLError> {
    let plan = UnifiedPlan::lower_with(statement, context.aggregates);
    register_plan(context, name, plan, &[], None)
}

pub fn register_plan(
    context: &PreparedRegistrationContext<'_>,
    name: String,
    logical_plan: UnifiedPlan,
    declared: &[ColumnType],
    source_sql: Option<&str>,
) -> Result<(), SQLError> {
    let _temporal_input = uqa_sql::expr::TemporalInputContext {
        transaction_clock_micros: context.transaction_timestamp_micros,
        date_order: context
            .date_order
            .unwrap_or_else(uqa_sql::expr::temporal_date_order),
    }
    .enter();
    let source_plan = Arc::new(logical_plan.clone());
    let definition = uqa_sql::prepared::definition::analyze_definition(
        &context.analysis,
        logical_plan,
        declared,
    )?;
    context.registry.write_definitions().insert(
        name,
        PreparedStatementPlan {
            source_plan,
            logical_plan: Arc::new(definition.logical_plan),
            needs_analysis: false,
            effective_search_path: definition.effective_search_path,
            dependencies: definition.dependencies,
            dependency_snapshot: definition.dependency_snapshot,
            plan: None,
            parameter_types: definition.parameter_types,
            result_schema: definition.result_schema,
            source_sql: source_sql.map(Arc::from),
            prepared_at_micros: (context.clock)(),
            from_sql: source_sql.is_some(),
            generic_plans: 0,
            custom_plans: 0,
            generic_cost: None,
            total_custom_cost: 0.0,
        },
    );
    Ok(())
}

#[cfg(test)]
mod tests;