uqa-engine 0.4.0

Engine: schema-aware table store, catalog restore, transactions
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

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

use super::{
    operator_execution_error, posting_list_to_scored, DriverResult, Engine, EngineDriver,
    OperatorOutput, OperatorTree, SQLError, SQLParam, ScalarExpr, ScoredEntry,
};

/// Bind, optimize and execute a predicate below its existing statement boundary.
pub fn run_optimised(
    engine: &Engine,
    table: &str,
    where_expr: Option<&ScalarExpr>,
    params: &[SQLParam],
) -> Result<Option<Vec<ScoredEntry>>, SQLError> {
    uqa_execution::query::block::where_filter::execute_typed_retrieval(
        &engine.source_execution_context(),
        table,
        where_expr,
        params,
        &crate::capabilities::query_scope::new_for_current_routine(engine),
    )
}

/// Optimise and execute an already-lowered tree through the same
/// planner/runtime boundary used by SQL `WHERE` lowering. Graph table
/// functions use this entry point too, so they do not maintain a
/// second physical dispatch implementation for nodes represented by
/// [`OperatorTree`].
pub(crate) fn execute_operator_tree(
    engine: &Engine,
    table: &str,
    signal_table: &str,
    params: &[SQLParam],
    tree: &OperatorTree,
) -> DriverResult<OperatorOutput> {
    engine.with_direct_query_snapshot(
        !tree_may_persist_calibration(tree),
        |engine| {
            engine
                .synchronize_catalog_registries()
                .map_err(|error| operator_execution_error("refresh operator catalog", error))?;
            execute_operator_tree_inner(engine, table, signal_table, params, tree)
        },
        std::convert::identity,
    )
}

/// A direct physical driver call owns one snapshot for the whole tree,
/// including graph-aware propagation and parallel child operators.
pub(super) fn execute_public_physical_node(
    driver: &EngineDriver<'_>,
    tree: &OperatorTree,
) -> DriverResult<OperatorOutput> {
    use super::OperatorTreeDriver as _;
    let engine = driver.engine;
    engine.with_direct_query_snapshot(
        !tree_may_persist_calibration(tree),
        |engine| {
            engine
                .synchronize_catalog_registries()
                .map_err(|error| operator_execution_error("refresh physical catalog", error))?;
            let scoped = engine
                .physical_retrieval_driver(driver.table, driver.table, driver.params)
                .with_parallel(driver.parallel.clone());
            scoped.execute_node(tree)
        },
        std::convert::identity,
    )
}

fn execute_operator_tree_inner(
    engine: &Engine,
    table: &str,
    signal_table: &str,
    params: &[SQLParam],
    tree: &OperatorTree,
) -> DriverResult<OperatorOutput> {
    uqa_execution::operator_tree::runtime::optimize_and_execute_tree(
        &engine.tree_execution_context(),
        table,
        signal_table,
        params,
        tree,
    )
}

/// Execute a concrete retrieval tree through the optimizer/plan-executor
/// boundary and convert its posting carrier for public engine APIs.
pub(crate) fn execute_scored_tree(
    engine: &Engine,
    table: &str,
    signal_table: &str,
    params: &[SQLParam],
    tree: &OperatorTree,
) -> DriverResult<Vec<ScoredEntry>> {
    let output = execute_operator_tree(engine, table, signal_table, params, tree)?;
    let posting = expect_posting_output(output, "retrieval API")?;
    Ok(posting_list_to_scored(&posting))
}

use uqa_execution::operator_tree::runtime::expect_posting_output;
use uqa_execution::operator_tree::runtime::tree_may_persist_calibration;

#[cfg(test)]
#[path = "transaction_boundary_tests.rs"]
mod transaction_boundary_tests;