uqa-engine 0.1.12

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

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

use super::*;
use uqa_execution::{Batch, ExecError, ExecResult, PhysicalOperator, RowSchema};
use uqa_planner::UnifiedPlan;

struct CloseOperator {
    schema: RowSchema,
    close_error: Option<&'static str>,
}

impl PhysicalOperator for CloseOperator {
    fn row_schema(&self) -> &RowSchema {
        &self.schema
    }

    fn open(&mut self) -> ExecResult<()> {
        unreachable!("the cleanup helper must not open the operator")
    }

    fn next(&mut self) -> ExecResult<Option<Batch>> {
        unreachable!("the cleanup helper must not pull the operator")
    }

    fn close(&mut self) -> ExecResult<()> {
        match self.close_error {
            Some(message) => Err(ExecError::Other(message.into())),
            None => Ok(()),
        }
    }
}

#[test]
fn physical_failure_preserves_the_original_error_when_close_succeeds() {
    let mut operator = CloseOperator {
        schema: RowSchema::new(Vec::new()),
        close_error: None,
    };
    let original = SQLError::TypeMismatch("primary".into());
    let original_message = original.to_string();
    let error = close_after_physical_failure(&mut operator, ExecError::SQL(original), "execution");
    assert_eq!(error.to_string(), original_message);
    assert!(matches!(error, SQLError::TypeMismatch(_)));
}

#[test]
fn physical_failure_reports_both_execution_and_close_errors() {
    let mut operator = CloseOperator {
        schema: RowSchema::new(Vec::new()),
        close_error: Some("cleanup"),
    };
    let error = close_after_physical_failure(
        &mut operator,
        ExecError::Other("primary".into()),
        "spill buffering",
    );
    let message = error.to_string();
    assert!(message.contains("primary"));
    assert!(message.contains("spill buffering"));
    assert!(message.contains("cleanup"));
}

#[test]
fn floating_limit_uses_postgresql_bigint_rounding_and_range_checks() {
    assert_eq!(float_limit_offset(42.0, "LIMIT").unwrap(), 42);
    assert_eq!(float_limit_offset(1.5, "LIMIT").unwrap(), 2);
    assert_eq!(float_limit_offset(2.5, "LIMIT").unwrap(), 2);
    assert_eq!(float_limit_offset(-0.5, "LIMIT").unwrap(), 0);
    for value in [f64::NAN, f64::INFINITY, -1.0, 9_223_372_036_854_775_808.0] {
        assert!(float_limit_offset(value, "LIMIT").is_err(), "{value}");
    }
}

#[test]
fn scalar_subquery_scope_restores_the_parent_arena_after_unwind() {
    let statement = uqa_sql::compile("SELECT 1")
        .unwrap()
        .into_iter()
        .next()
        .unwrap();
    let UnifiedPlan::Query(query) = UnifiedPlan::lower(statement) else {
        panic!("SELECT must lower to a query plan");
    };
    let query = *query;
    let mut scope = CteScope::new();
    scope.scalar_subqueries.push(query.clone());

    let unwind = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        let _guard = scope.enter_scalar_subqueries(&[query.clone(), query]);
        panic!("exercise scalar-subquery scope cleanup");
    }));

    assert!(unwind.is_err());
    assert_eq!(scope.scalar_subqueries.len(), 1);
}