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
101
102
103
104
105
106

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

//! Combine repeatable query outputs through bounded set operations and ordering.

use super::{limit::attach_order_limit, RelationalContext};
use crate::query::{
    consumer::QueryOutputMode,
    ordering::identity_order_columns,
    output::QueryOutput,
    projection::{physical_exec_error, physical_work_mem_bytes},
    CteScope,
};
use uqa_sql::{ast::SetOpKind, plan::QueryBlockPlan, SQLError, SQLParam};

pub struct SetSpillExecution<'a> {
    kind: SetOpKind,
    all: bool,
    columns: Vec<String>,
    lhs: crate::SharedSpill,
    rhs: crate::SharedSpill,
    order_plan: Option<&'a QueryBlockPlan>,
    output_mode: QueryOutputMode<'a>,
}

impl<'a> SetSpillExecution<'a> {
    pub fn new(
        kind: SetOpKind,
        all: bool,
        columns: Vec<String>,
        lhs: crate::SharedSpill,
        rhs: crate::SharedSpill,
        order_plan: Option<&'a QueryBlockPlan>,
        output_mode: QueryOutputMode<'a>,
    ) -> Self {
        Self {
            kind,
            all,
            columns,
            lhs,
            rhs,
            order_plan,
            output_mode,
        }
    }
}

pub fn combine_set_spills_with_order_output<'a, S: Clone + 'static>(
    context: RelationalContext<'a, S>,
    execution: SetSpillExecution<'a>,
    params: &'a [SQLParam],
    ctes: &CteScope<S>,
) -> Result<QueryOutput, SQLError> {
    use crate::{ExternalSetOperation, PhysicalOperator};

    let public_positions = || {
        execution
            .columns
            .iter()
            .cloned()
            .enumerate()
            .map(|(position, column)| (column, position))
            .collect::<Vec<_>>()
    };
    let left: Box<dyn PhysicalOperator> = Box::new(crate::ColumnSelection::with_positions(
        Box::new(crate::SharedSpillScan::new(execution.lhs)),
        public_positions(),
    ));
    let right: Box<dyn PhysicalOperator> = Box::new(crate::ColumnSelection::with_positions(
        Box::new(crate::SharedSpillScan::new(execution.rhs)),
        public_positions(),
    ));
    let mut operation: Box<dyn PhysicalOperator + '_> = Box::new(
        ExternalSetOperation::new(
            left,
            right,
            execution.kind,
            execution.all,
            physical_work_mem_bytes(context.runtime)?,
        )
        .map_err(physical_exec_error)?,
    );
    if let Some(order_plan) = execution.order_plan {
        let output = identity_order_columns(&execution.columns);
        operation = attach_order_limit(
            operation,
            order_plan,
            &output,
            context,
            params,
            ctes,
            context.runtime,
            context.evaluator(params, ctes),
            None,
        )?;
    }
    crate::query::collection::collect_query_operator(
        context.runtime,
        execution.columns,
        operation,
        execution.output_mode,
    )
}