iridium_core 0.1.12

SQL Server-compatible Rust engine core for Iridium SQL
Documentation 
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

use parking_lot::Mutex;
use std::collections::HashSet;

use super::{EngineCatalog, EngineStorage};
use crate::error::DbError;
use crate::executor::database::SessionManagerService;
use crate::executor::journal::Journal;
use crate::executor::locks::SessionId;
use crate::executor::session::{SessionManager, SessionRuntime};
use crate::executor::string_norm::normalize_identifier;

impl<C, S> SessionManager for SessionManagerService<C, S>
where
    C: EngineCatalog,
    S: EngineStorage,
{
    fn create_session(&self) -> SessionId {
        let id = self
            .state
            .next_session_id
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        self.state
            .sessions
            .insert(id, Mutex::new(SessionRuntime::new()));
        self.state.deadlock_priorities.insert(id, 0);
        id
    }

    fn reset_session(&self, session_id: SessionId) -> Result<(), DbError> {
        let session_mutex = self
            .state
            .sessions
            .get(&session_id)
            .ok_or_else(|| DbError::Execution(format!("session {} not found", session_id)))?;
        let mut session = session_mutex.lock();
        let mut physical_tables = HashSet::new();
        for table in session.tables.temp_map.values() {
            physical_tables.insert(table.clone());
        }
        for table in session.tables.var_map.values() {
            physical_tables.insert(table.clone());
        }
        session.reset();
        self.state
            .deadlock_priorities
            .insert(session_id, session.options.deadlock_priority);
        drop(session);

        self.state
            .table_locks
            .lock()
            .release_all_for_session(session_id);

        if !physical_tables.is_empty() {
            let mut storage = self.state.storage.write();
            for table_name in physical_tables {
                let Some(table) = storage
                    .catalog
                    .get_tables()
                    .iter()
                    .find(|table| table.name.eq_ignore_ascii_case(&table_name))
                    .cloned()
                else {
                    continue;
                };

                let schema_name = table.schema_name.clone();
                let _ = storage.catalog.drop_table(&schema_name, &table_name);
                storage.storage.remove_table(table.id)?;
                storage.table_versions.remove(&format!(
                    "{}.{}",
                    normalize_identifier(&schema_name),
                    normalize_identifier(&table_name)
                ));
            }
        }
        Ok(())
    }

    fn close_session(&self, session_id: SessionId) -> Result<(), DbError> {
        self.state
            .table_locks
            .lock()
            .release_all_for_session(session_id);
        self.state.deadlock_priorities.remove(&session_id);
        let removed = self.state.sessions.remove(&session_id);
        if removed.is_none() {
            return Err(DbError::Execution(format!(
                "session {} not found",
                session_id
            )));
        }
        Ok(())
    }

    fn set_session_journal(
        &self,
        session_id: SessionId,
        journal: Box<dyn Journal>,
    ) -> Result<(), DbError> {
        let session_mutex = self
            .state
            .sessions
            .get(&session_id)
            .ok_or_else(|| DbError::Execution(format!("session {} not found", session_id)))?;
        let mut session = session_mutex.lock();
        session.journal = journal;
        Ok(())
    }
}