uqa-execution 0.4.5

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

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

//! Live session cursor descriptions independent of the executor's current location.

use parking_lot::Mutex;
use std::{
    collections::BTreeMap,
    sync::{Arc, Weak},
};
use uqa_sql::{catalog::session::CursorMetadata, SQLError};

#[derive(Default)]
struct RegistryState {
    entries: BTreeMap<String, Weak<CursorMetadata>>,
    next_id: u64,
}

/// The session retains this registry; an operational portal retains its registration.
#[derive(Default)]
pub struct PortalRegistry {
    state: Arc<Mutex<RegistryState>>,
}

/// Moving an executor out of its session map does not hide or release its cursor name.
pub struct PortalRegistration {
    metadata: Arc<CursorMetadata>,
    registry: Weak<Mutex<RegistryState>>,
}

impl PortalRegistry {
    pub fn ensure_available(&self, name: &str) -> Result<(), SQLError> {
        ensure_available(&self.state.lock(), name)
    }

    pub fn register(&self, metadata: CursorMetadata) -> Result<PortalRegistration, SQLError> {
        let mut state = self.state.lock();
        ensure_available(&state, &metadata.name)?;
        let metadata = Arc::new(metadata);
        state
            .entries
            .insert(metadata.name.clone(), Arc::downgrade(&metadata));
        Ok(PortalRegistration {
            metadata,
            registry: Arc::downgrade(&self.state),
        })
    }

    pub fn snapshot(&self) -> Vec<CursorMetadata> {
        self.state
            .lock()
            .entries
            .values()
            .filter_map(Weak::upgrade)
            .map(|metadata| (*metadata).clone())
            .collect()
    }

    pub fn allocate_name(&self) -> String {
        let mut state = self.state.lock();
        loop {
            state.next_id = state.next_id.wrapping_add(1);
            let name = format!("<unnamed portal {}>", state.next_id);
            if ensure_available(&state, &name).is_ok() {
                return name;
            }
        }
    }
}

fn ensure_available(state: &RegistryState, name: &str) -> Result<(), SQLError> {
    if state
        .entries
        .get(name)
        .is_some_and(|entry| entry.strong_count() != 0)
    {
        return Err(SQLError::Routine {
            sqlstate: "42P03".into(),
            message: format!("cursor \"{name}\" already exists"),
        });
    }
    Ok(())
}

impl Drop for PortalRegistration {
    fn drop(&mut self) {
        if let Some(registry) = self.registry.upgrade() {
            let mut state = registry.lock();
            if state
                .entries
                .get(&self.metadata.name)
                .is_some_and(|entry| entry.ptr_eq(&Arc::downgrade(&self.metadata)))
            {
                state.entries.remove(&self.metadata.name);
            }
        }
    }
}

#[cfg(test)]
mod tests;