vibesql_executor/
cursor.rs

1//! Cursor execution (SQL:1999 Feature E121)
2//!
3//! This module provides execution support for SQL cursor operations:
4//! - DECLARE CURSOR: Declare a cursor for a query
5//! - OPEN cursor: Execute the query and materialize results
6//! - FETCH: Retrieve rows from an open cursor
7//! - CLOSE cursor: Close and release cursor resources
8//!
9//! Cursors provide a way to iterate over query results row by row, which is
10//! useful for processing large result sets without loading everything into memory.
11
12use std::collections::HashMap;
13
14use vibesql_ast::{
15    CloseCursorStmt, DeclareCursorStmt, FetchOrientation, FetchStmt, OpenCursorStmt, SelectStmt,
16};
17use vibesql_storage::{Database, Row};
18
19use crate::{errors::ExecutorError, SelectExecutor};
20
21/// A cursor holding query definition and materialized results
22#[derive(Debug, Clone)]
23pub struct Cursor {
24    /// Cursor name
25    pub name: String,
26    /// The SELECT query for this cursor
27    pub query: Box<SelectStmt>,
28    /// Materialized results (populated on OPEN)
29    pub result: Option<CursorResult>,
30    /// Current position (0 = before first row, 1 = first row, etc.)
31    pub position: usize,
32    /// Whether cursor supports backward movement
33    pub scroll: bool,
34    /// Whether cursor survives transaction commit (WITH HOLD)
35    pub holdable: bool,
36    /// Whether cursor uses snapshot isolation (INSENSITIVE)
37    pub insensitive: bool,
38}
39
40/// Materialized cursor result set
41#[derive(Debug, Clone)]
42pub struct CursorResult {
43    /// Column names
44    pub columns: Vec<String>,
45    /// All rows from the query
46    pub rows: Vec<Row>,
47}
48
49/// Result of a FETCH operation
50#[derive(Debug, Clone)]
51pub struct FetchResult {
52    /// Column names
53    pub columns: Vec<String>,
54    /// Fetched rows (0 or 1 for single-row fetch, potentially more for FETCH ALL)
55    pub rows: Vec<Row>,
56}
57
58impl FetchResult {
59    /// Create an empty fetch result (no row at current position)
60    pub fn empty(columns: Vec<String>) -> Self {
61        Self { columns, rows: vec![] }
62    }
63
64    /// Create a fetch result with a single row
65    pub fn single(columns: Vec<String>, row: Row) -> Self {
66        Self { columns, rows: vec![row] }
67    }
68}
69
70/// Storage for cursors within a session
71#[derive(Debug, Default)]
72pub struct CursorStore {
73    cursors: HashMap<String, Cursor>,
74}
75
76impl CursorStore {
77    /// Create a new empty cursor store
78    pub fn new() -> Self {
79        Self { cursors: HashMap::new() }
80    }
81
82    /// Declare a new cursor
83    pub fn declare(&mut self, stmt: &DeclareCursorStmt) -> Result<(), ExecutorError> {
84        let name = stmt.cursor_name.to_uppercase();
85
86        if self.cursors.contains_key(&name) {
87            return Err(ExecutorError::CursorAlreadyExists(name));
88        }
89
90        let cursor = Cursor {
91            name: name.clone(),
92            query: stmt.query.clone(),
93            result: None,
94            position: 0,
95            scroll: stmt.scroll,
96            holdable: stmt.hold.unwrap_or(false),
97            insensitive: stmt.insensitive,
98        };
99
100        self.cursors.insert(name, cursor);
101        Ok(())
102    }
103
104    /// Open a cursor (execute query and materialize results)
105    pub fn open(&mut self, stmt: &OpenCursorStmt, db: &Database) -> Result<(), ExecutorError> {
106        let name = stmt.cursor_name.to_uppercase();
107
108        let cursor = self
109            .cursors
110            .get_mut(&name)
111            .ok_or_else(|| ExecutorError::CursorNotFound(name.clone()))?;
112
113        if cursor.result.is_some() {
114            return Err(ExecutorError::CursorAlreadyOpen(name));
115        }
116
117        // Execute the query
118        let executor = SelectExecutor::new(db);
119        let select_result = executor.execute_with_columns(&cursor.query)?;
120
121        cursor.result =
122            Some(CursorResult { columns: select_result.columns, rows: select_result.rows });
123        cursor.position = 0; // Before first row
124
125        Ok(())
126    }
127
128    /// Fetch from a cursor
129    pub fn fetch(&mut self, stmt: &FetchStmt) -> Result<FetchResult, ExecutorError> {
130        let name = stmt.cursor_name.to_uppercase();
131
132        let cursor = self
133            .cursors
134            .get_mut(&name)
135            .ok_or_else(|| ExecutorError::CursorNotFound(name.clone()))?;
136
137        let result =
138            cursor.result.as_ref().ok_or_else(|| ExecutorError::CursorNotOpen(name.clone()))?;
139
140        let row_count = result.rows.len();
141
142        // Calculate new position based on fetch orientation
143        let new_position = match &stmt.orientation {
144            FetchOrientation::Next => cursor.position.saturating_add(1),
145            FetchOrientation::Prior => {
146                if !cursor.scroll {
147                    return Err(ExecutorError::CursorNotScrollable(name));
148                }
149                cursor.position.saturating_sub(1)
150            }
151            FetchOrientation::First => {
152                if !cursor.scroll && cursor.position > 0 {
153                    return Err(ExecutorError::CursorNotScrollable(name));
154                }
155                1
156            }
157            FetchOrientation::Last => {
158                if !cursor.scroll {
159                    return Err(ExecutorError::CursorNotScrollable(name));
160                }
161                row_count
162            }
163            FetchOrientation::Absolute(n) => {
164                if !cursor.scroll {
165                    return Err(ExecutorError::CursorNotScrollable(name));
166                }
167                if *n >= 0 {
168                    *n as usize
169                } else {
170                    // Negative: count from end
171                    row_count.saturating_sub((-*n - 1) as usize)
172                }
173            }
174            FetchOrientation::Relative(n) => {
175                if !cursor.scroll && *n < 0 {
176                    return Err(ExecutorError::CursorNotScrollable(name));
177                }
178                if *n >= 0 {
179                    cursor.position.saturating_add(*n as usize)
180                } else {
181                    cursor.position.saturating_sub((-*n) as usize)
182                }
183            }
184        };
185
186        cursor.position = new_position;
187
188        // Return row at position (1-indexed, 0 = before first)
189        if new_position > 0 && new_position <= row_count {
190            Ok(FetchResult::single(result.columns.clone(), result.rows[new_position - 1].clone()))
191        } else {
192            // No row at this position
193            Ok(FetchResult::empty(result.columns.clone()))
194        }
195    }
196
197    /// Close a cursor
198    pub fn close(&mut self, stmt: &CloseCursorStmt) -> Result<(), ExecutorError> {
199        let name = stmt.cursor_name.to_uppercase();
200
201        if self.cursors.remove(&name).is_none() {
202            return Err(ExecutorError::CursorNotFound(name));
203        }
204
205        Ok(())
206    }
207
208    /// Check if a cursor exists
209    pub fn exists(&self, name: &str) -> bool {
210        self.cursors.contains_key(&name.to_uppercase())
211    }
212
213    /// Check if a cursor is open
214    pub fn is_open(&self, name: &str) -> bool {
215        self.cursors.get(&name.to_uppercase()).map(|c| c.result.is_some()).unwrap_or(false)
216    }
217
218    /// Get the number of declared cursors
219    pub fn count(&self) -> usize {
220        self.cursors.len()
221    }
222
223    /// Clear all non-holdable cursors (called on transaction commit/rollback)
224    pub fn clear_non_holdable(&mut self) {
225        self.cursors.retain(|_, cursor| cursor.holdable);
226    }
227
228    /// Clear all cursors
229    pub fn clear(&mut self) {
230        self.cursors.clear();
231    }
232}
233
234/// Executor for cursor operations
235pub struct CursorExecutor;
236
237impl CursorExecutor {
238    /// Execute DECLARE CURSOR statement
239    pub fn declare(store: &mut CursorStore, stmt: &DeclareCursorStmt) -> Result<(), ExecutorError> {
240        store.declare(stmt)
241    }
242
243    /// Execute OPEN CURSOR statement
244    pub fn open(
245        store: &mut CursorStore,
246        stmt: &OpenCursorStmt,
247        db: &Database,
248    ) -> Result<(), ExecutorError> {
249        store.open(stmt, db)
250    }
251
252    /// Execute FETCH statement
253    pub fn fetch(store: &mut CursorStore, stmt: &FetchStmt) -> Result<FetchResult, ExecutorError> {
254        store.fetch(stmt)
255    }
256
257    /// Execute CLOSE CURSOR statement
258    pub fn close(store: &mut CursorStore, stmt: &CloseCursorStmt) -> Result<(), ExecutorError> {
259        store.close(stmt)
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use vibesql_ast::{
266        CloseCursorStmt, CursorUpdatability, DeclareCursorStmt, FetchOrientation, FetchStmt,
267        FromClause, OpenCursorStmt, SelectItem, SelectStmt,
268    };
269    use vibesql_catalog::{ColumnSchema, TableSchema};
270    use vibesql_types::{DataType, SqlValue};
271
272    use super::*;
273
274    fn create_test_db() -> Database {
275        let mut db = Database::new();
276        db.catalog.set_case_sensitive_identifiers(false);
277
278        // Create employees table
279        let columns = vec![
280            ColumnSchema::new("id".to_string(), DataType::Integer, false),
281            ColumnSchema::new(
282                "name".to_string(),
283                DataType::Varchar { max_length: Some(100) },
284                true,
285            ),
286            ColumnSchema::new("salary".to_string(), DataType::Integer, true),
287        ];
288        let schema =
289            TableSchema::with_primary_key("employees".to_string(), columns, vec!["id".to_string()]);
290        db.create_table(schema).unwrap();
291
292        // Insert test data
293        db.insert_row(
294            "employees",
295            Row::new(vec![
296                SqlValue::Integer(1),
297                SqlValue::Varchar(arcstr::ArcStr::from("Alice")),
298                SqlValue::Integer(50000),
299            ]),
300        )
301        .unwrap();
302        db.insert_row(
303            "employees",
304            Row::new(vec![
305                SqlValue::Integer(2),
306                SqlValue::Varchar(arcstr::ArcStr::from("Bob")),
307                SqlValue::Integer(60000),
308            ]),
309        )
310        .unwrap();
311        db.insert_row(
312            "employees",
313            Row::new(vec![
314                SqlValue::Integer(3),
315                SqlValue::Varchar(arcstr::ArcStr::from("Carol")),
316                SqlValue::Integer(55000),
317            ]),
318        )
319        .unwrap();
320
321        db
322    }
323
324    fn create_select_stmt() -> SelectStmt {
325        SelectStmt {
326            with_clause: None,
327            distinct: false,
328            select_list: vec![SelectItem::Wildcard { alias: None }],
329            into_table: None,
330            into_variables: None,
331            from: Some(FromClause::Table {
332                index_hint: None,
333                name: "employees".to_string(),
334                alias: None,
335                column_aliases: None,
336                quoted: false,
337            }),
338            where_clause: None,
339            group_by: None,
340            having: None,
341            window_definitions: None,
342            order_by: None,
343            limit: None,
344            offset: None,
345            set_operation: None,
346            values: None,
347        }
348    }
349
350    #[test]
351    fn test_declare_cursor() {
352        let mut store = CursorStore::new();
353
354        let stmt = DeclareCursorStmt {
355            cursor_name: "emp_cursor".to_string(),
356            insensitive: false,
357            scroll: false,
358            hold: None,
359            query: Box::new(create_select_stmt()),
360            updatability: CursorUpdatability::Unspecified,
361        };
362
363        assert!(store.declare(&stmt).is_ok());
364        assert!(store.exists("emp_cursor"));
365        assert!(!store.is_open("emp_cursor"));
366    }
367
368    #[test]
369    fn test_declare_cursor_already_exists() {
370        let mut store = CursorStore::new();
371
372        let stmt = DeclareCursorStmt {
373            cursor_name: "emp_cursor".to_string(),
374            insensitive: false,
375            scroll: false,
376            hold: None,
377            query: Box::new(create_select_stmt()),
378            updatability: CursorUpdatability::Unspecified,
379        };
380
381        store.declare(&stmt).unwrap();
382        let result = store.declare(&stmt);
383        assert!(matches!(result, Err(ExecutorError::CursorAlreadyExists(_))));
384    }
385
386    #[test]
387    fn test_open_cursor() {
388        let db = create_test_db();
389        let mut store = CursorStore::new();
390
391        let declare_stmt = DeclareCursorStmt {
392            cursor_name: "emp_cursor".to_string(),
393            insensitive: false,
394            scroll: false,
395            hold: None,
396            query: Box::new(create_select_stmt()),
397            updatability: CursorUpdatability::Unspecified,
398        };
399        store.declare(&declare_stmt).unwrap();
400
401        let open_stmt = OpenCursorStmt { cursor_name: "emp_cursor".to_string() };
402        assert!(store.open(&open_stmt, &db).is_ok());
403        assert!(store.is_open("emp_cursor"));
404    }
405
406    #[test]
407    fn test_open_cursor_not_found() {
408        let db = create_test_db();
409        let mut store = CursorStore::new();
410
411        let open_stmt = OpenCursorStmt { cursor_name: "nonexistent".to_string() };
412        let result = store.open(&open_stmt, &db);
413        assert!(matches!(result, Err(ExecutorError::CursorNotFound(_))));
414    }
415
416    #[test]
417    fn test_open_cursor_already_open() {
418        let db = create_test_db();
419        let mut store = CursorStore::new();
420
421        let declare_stmt = DeclareCursorStmt {
422            cursor_name: "emp_cursor".to_string(),
423            insensitive: false,
424            scroll: false,
425            hold: None,
426            query: Box::new(create_select_stmt()),
427            updatability: CursorUpdatability::Unspecified,
428        };
429        store.declare(&declare_stmt).unwrap();
430
431        let open_stmt = OpenCursorStmt { cursor_name: "emp_cursor".to_string() };
432        store.open(&open_stmt, &db).unwrap();
433
434        let result = store.open(&open_stmt, &db);
435        assert!(matches!(result, Err(ExecutorError::CursorAlreadyOpen(_))));
436    }
437
438    #[test]
439    fn test_fetch_next() {
440        let db = create_test_db();
441        let mut store = CursorStore::new();
442
443        let declare_stmt = DeclareCursorStmt {
444            cursor_name: "emp_cursor".to_string(),
445            insensitive: false,
446            scroll: false,
447            hold: None,
448            query: Box::new(create_select_stmt()),
449            updatability: CursorUpdatability::Unspecified,
450        };
451        store.declare(&declare_stmt).unwrap();
452
453        let open_stmt = OpenCursorStmt { cursor_name: "emp_cursor".to_string() };
454        store.open(&open_stmt, &db).unwrap();
455
456        // Fetch first row
457        let fetch_stmt = FetchStmt {
458            cursor_name: "emp_cursor".to_string(),
459            orientation: FetchOrientation::Next,
460            into_variables: None,
461        };
462        let result = store.fetch(&fetch_stmt).unwrap();
463        assert_eq!(result.rows.len(), 1);
464        assert_eq!(result.rows[0].values[0], SqlValue::Integer(1));
465
466        // Fetch second row
467        let result = store.fetch(&fetch_stmt).unwrap();
468        assert_eq!(result.rows.len(), 1);
469        assert_eq!(result.rows[0].values[0], SqlValue::Integer(2));
470
471        // Fetch third row
472        let result = store.fetch(&fetch_stmt).unwrap();
473        assert_eq!(result.rows.len(), 1);
474        assert_eq!(result.rows[0].values[0], SqlValue::Integer(3));
475
476        // Fetch past end - should return empty
477        let result = store.fetch(&fetch_stmt).unwrap();
478        assert_eq!(result.rows.len(), 0);
479    }
480
481    #[test]
482    fn test_fetch_from_unopened_cursor() {
483        let mut store = CursorStore::new();
484
485        let declare_stmt = DeclareCursorStmt {
486            cursor_name: "emp_cursor".to_string(),
487            insensitive: false,
488            scroll: false,
489            hold: None,
490            query: Box::new(create_select_stmt()),
491            updatability: CursorUpdatability::Unspecified,
492        };
493        store.declare(&declare_stmt).unwrap();
494
495        let fetch_stmt = FetchStmt {
496            cursor_name: "emp_cursor".to_string(),
497            orientation: FetchOrientation::Next,
498            into_variables: None,
499        };
500        let result = store.fetch(&fetch_stmt);
501        assert!(matches!(result, Err(ExecutorError::CursorNotOpen(_))));
502    }
503
504    #[test]
505    fn test_fetch_prior_non_scrollable() {
506        let db = create_test_db();
507        let mut store = CursorStore::new();
508
509        let declare_stmt = DeclareCursorStmt {
510            cursor_name: "emp_cursor".to_string(),
511            insensitive: false,
512            scroll: false, // Non-scrollable
513            hold: None,
514            query: Box::new(create_select_stmt()),
515            updatability: CursorUpdatability::Unspecified,
516        };
517        store.declare(&declare_stmt).unwrap();
518
519        let open_stmt = OpenCursorStmt { cursor_name: "emp_cursor".to_string() };
520        store.open(&open_stmt, &db).unwrap();
521
522        // Move to first row
523        let fetch_next = FetchStmt {
524            cursor_name: "emp_cursor".to_string(),
525            orientation: FetchOrientation::Next,
526            into_variables: None,
527        };
528        store.fetch(&fetch_next).unwrap();
529
530        // Try PRIOR on non-scrollable cursor
531        let fetch_prior = FetchStmt {
532            cursor_name: "emp_cursor".to_string(),
533            orientation: FetchOrientation::Prior,
534            into_variables: None,
535        };
536        let result = store.fetch(&fetch_prior);
537        assert!(matches!(result, Err(ExecutorError::CursorNotScrollable(_))));
538    }
539
540    #[test]
541    fn test_scroll_cursor() {
542        let db = create_test_db();
543        let mut store = CursorStore::new();
544
545        let declare_stmt = DeclareCursorStmt {
546            cursor_name: "scroll_cursor".to_string(),
547            insensitive: false,
548            scroll: true, // Scrollable
549            hold: None,
550            query: Box::new(create_select_stmt()),
551            updatability: CursorUpdatability::Unspecified,
552        };
553        store.declare(&declare_stmt).unwrap();
554
555        let open_stmt = OpenCursorStmt { cursor_name: "scroll_cursor".to_string() };
556        store.open(&open_stmt, &db).unwrap();
557
558        // Fetch LAST
559        let fetch_last = FetchStmt {
560            cursor_name: "scroll_cursor".to_string(),
561            orientation: FetchOrientation::Last,
562            into_variables: None,
563        };
564        let result = store.fetch(&fetch_last).unwrap();
565        assert_eq!(result.rows.len(), 1);
566        assert_eq!(result.rows[0].values[0], SqlValue::Integer(3));
567
568        // Fetch FIRST
569        let fetch_first = FetchStmt {
570            cursor_name: "scroll_cursor".to_string(),
571            orientation: FetchOrientation::First,
572            into_variables: None,
573        };
574        let result = store.fetch(&fetch_first).unwrap();
575        assert_eq!(result.rows.len(), 1);
576        assert_eq!(result.rows[0].values[0], SqlValue::Integer(1));
577
578        // Fetch ABSOLUTE 2
579        let fetch_abs = FetchStmt {
580            cursor_name: "scroll_cursor".to_string(),
581            orientation: FetchOrientation::Absolute(2),
582            into_variables: None,
583        };
584        let result = store.fetch(&fetch_abs).unwrap();
585        assert_eq!(result.rows.len(), 1);
586        assert_eq!(result.rows[0].values[0], SqlValue::Integer(2));
587    }
588
589    #[test]
590    fn test_close_cursor() {
591        let db = create_test_db();
592        let mut store = CursorStore::new();
593
594        let declare_stmt = DeclareCursorStmt {
595            cursor_name: "emp_cursor".to_string(),
596            insensitive: false,
597            scroll: false,
598            hold: None,
599            query: Box::new(create_select_stmt()),
600            updatability: CursorUpdatability::Unspecified,
601        };
602        store.declare(&declare_stmt).unwrap();
603
604        let open_stmt = OpenCursorStmt { cursor_name: "emp_cursor".to_string() };
605        store.open(&open_stmt, &db).unwrap();
606
607        let close_stmt = CloseCursorStmt { cursor_name: "emp_cursor".to_string() };
608        assert!(store.close(&close_stmt).is_ok());
609        assert!(!store.exists("emp_cursor"));
610    }
611
612    #[test]
613    fn test_close_nonexistent_cursor() {
614        let mut store = CursorStore::new();
615
616        let close_stmt = CloseCursorStmt { cursor_name: "nonexistent".to_string() };
617        let result = store.close(&close_stmt);
618        assert!(matches!(result, Err(ExecutorError::CursorNotFound(_))));
619    }
620
621    #[test]
622    fn test_case_insensitive_cursor_names() {
623        let db = create_test_db();
624        let mut store = CursorStore::new();
625
626        let declare_stmt = DeclareCursorStmt {
627            cursor_name: "My_Cursor".to_string(),
628            insensitive: false,
629            scroll: false,
630            hold: None,
631            query: Box::new(create_select_stmt()),
632            updatability: CursorUpdatability::Unspecified,
633        };
634        store.declare(&declare_stmt).unwrap();
635
636        // Open with different case
637        let open_stmt = OpenCursorStmt { cursor_name: "MY_CURSOR".to_string() };
638        assert!(store.open(&open_stmt, &db).is_ok());
639
640        // Close with different case
641        let close_stmt = CloseCursorStmt { cursor_name: "my_cursor".to_string() };
642        assert!(store.close(&close_stmt).is_ok());
643    }
644
645    #[test]
646    fn test_holdable_cursor() {
647        let mut store = CursorStore::new();
648
649        // Create holdable cursor
650        let holdable_stmt = DeclareCursorStmt {
651            cursor_name: "holdable".to_string(),
652            insensitive: false,
653            scroll: false,
654            hold: Some(true),
655            query: Box::new(create_select_stmt()),
656            updatability: CursorUpdatability::Unspecified,
657        };
658        store.declare(&holdable_stmt).unwrap();
659
660        // Create non-holdable cursor
661        let non_holdable_stmt = DeclareCursorStmt {
662            cursor_name: "non_holdable".to_string(),
663            insensitive: false,
664            scroll: false,
665            hold: Some(false),
666            query: Box::new(create_select_stmt()),
667            updatability: CursorUpdatability::Unspecified,
668        };
669        store.declare(&non_holdable_stmt).unwrap();
670
671        assert_eq!(store.count(), 2);
672
673        // Clear non-holdable cursors (simulating transaction commit)
674        store.clear_non_holdable();
675
676        assert_eq!(store.count(), 1);
677        assert!(store.exists("holdable"));
678        assert!(!store.exists("non_holdable"));
679    }
680}
vibesql_executor/cursor.rs

vibesql_executor/
cursor.rs
