alopex_embedded/
sql_api.rs

1use alopex_core::kv::KVStore;
2use alopex_core::types::TxnMode;
3use alopex_core::KVTransaction;
4use alopex_sql::catalog::CatalogOverlay;
5use alopex_sql::catalog::TxnCatalogView;
6use alopex_sql::executor::query::execute_query_streaming;
7use alopex_sql::executor::query::iterator::VecIterator;
8use alopex_sql::executor::query::RowIterator;
9use alopex_sql::executor::{
10    build_streaming_pipeline, ColumnInfo, ExecutionResult, Executor, QueryRowIterator, Row,
11};
12use alopex_sql::planner::typed_expr::Projection;
13use alopex_sql::storage::{SqlValue, TxnBridge};
14use alopex_sql::AlopexDialect;
15use alopex_sql::Parser;
16use alopex_sql::Planner;
17use alopex_sql::Statement;
18use alopex_sql::StatementKind;
19
20use crate::Database;
21use crate::Error;
22use crate::Result;
23use crate::SqlResult;
24use crate::Transaction;
25
26/// Streaming row access for FR-7 compliance.
27///
28/// This struct provides access to query results in a streaming fashion,
29/// where the transaction is kept alive for the duration of row iteration.
30/// The lifetime `'a` is tied to the transaction scope.
31pub struct StreamingRows<'a> {
32    columns: Vec<ColumnInfo>,
33    iter: Box<dyn RowIterator + 'a>,
34    projection: Projection,
35    schema: Vec<alopex_sql::catalog::ColumnMetadata>,
36}
37
38impl<'a> StreamingRows<'a> {
39    /// Get column information for the query result.
40    pub fn columns(&self) -> &[ColumnInfo] {
41        &self.columns
42    }
43
44    /// Fetch the next row, returning `None` when exhausted.
45    ///
46    /// Rows are fetched on-demand from storage, enabling true streaming.
47    pub fn next_row(&mut self) -> Result<Option<Vec<SqlValue>>> {
48        match self.iter.next_row() {
49            Some(result) => {
50                let row = result.map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
51                let projected = self.project_row(&row)?;
52                Ok(Some(projected))
53            }
54            None => Ok(None),
55        }
56    }
57
58    /// Apply projection to a row.
59    fn project_row(&self, row: &Row) -> Result<Vec<SqlValue>> {
60        match &self.projection {
61            Projection::All(names) => {
62                // Return values in the order specified by names
63                let mut result = Vec::with_capacity(names.len());
64                for name in names {
65                    let idx = self
66                        .schema
67                        .iter()
68                        .position(|c| &c.name == name)
69                        .ok_or_else(|| {
70                            Error::Sql(alopex_sql::SqlError::Execution {
71                                message: format!("column not found: {}", name),
72                                code: "ALOPEX-E020",
73                            })
74                        })?;
75                    result.push(row.values.get(idx).cloned().unwrap_or(SqlValue::Null));
76                }
77                Ok(result)
78            }
79            Projection::Columns(cols) => {
80                use alopex_sql::executor::evaluator::{evaluate, EvalContext};
81                let ctx = EvalContext::new(&row.values);
82                let mut result = Vec::with_capacity(cols.len());
83                for col in cols {
84                    let value = evaluate(&col.expr, &ctx)
85                        .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
86                    result.push(value);
87                }
88                Ok(result)
89            }
90        }
91    }
92}
93
94/// Result type for callback-based streaming query.
95pub enum StreamingQueryResult<R> {
96    /// DDL operation success.
97    Success,
98    /// DML operation with affected row count.
99    RowsAffected(u64),
100    /// Query result processed by callback.
101    QueryProcessed(R),
102}
103
104/// Streaming SQL execution result for FR-7 compliance.
105///
106/// This enum enables true streaming output for SELECT queries by returning
107/// an iterator instead of a materialized Vec.
108pub enum SqlStreamingResult {
109    /// DDL operation success (CREATE/DROP TABLE/INDEX).
110    Success,
111    /// DML operation success with affected row count.
112    RowsAffected(u64),
113    /// Query result with streaming row iterator.
114    Query(QueryRowIterator<'static>),
115}
116
117fn parse_sql(sql: &str) -> Result<Vec<Statement>> {
118    let dialect = AlopexDialect;
119    Parser::parse_sql(&dialect, sql).map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))
120}
121
122fn stmt_requires_write(stmt: &Statement) -> bool {
123    !matches!(stmt.kind, StatementKind::Select(_))
124}
125
126fn stmt_changes_catalog(stmt: &Statement) -> bool {
127    matches!(
128        stmt.kind,
129        StatementKind::CreateTable(_)
130            | StatementKind::DropTable(_)
131            | StatementKind::CreateIndex(_)
132            | StatementKind::DropIndex(_)
133    )
134}
135
136fn plan_stmt<'a, S: KVStore>(
137    catalog: &'a alopex_sql::catalog::PersistentCatalog<S>,
138    overlay: &'a CatalogOverlay,
139    stmt: &Statement,
140) -> Result<alopex_sql::LogicalPlan> {
141    let view = TxnCatalogView::new(catalog, overlay);
142    let planner = Planner::new(&view);
143    planner
144        .plan(stmt)
145        .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))
146}
147
148/// Build column info from projection and schema.
149fn build_column_info(
150    projection: &Projection,
151    schema: &[alopex_sql::catalog::ColumnMetadata],
152) -> Result<Vec<ColumnInfo>> {
153    match projection {
154        Projection::All(names) => {
155            let mut cols = Vec::with_capacity(names.len());
156            for name in names {
157                let meta = schema.iter().find(|c| &c.name == name).ok_or_else(|| {
158                    Error::Sql(alopex_sql::SqlError::Execution {
159                        message: format!("column not found: {}", name),
160                        code: "ALOPEX-E020",
161                    })
162                })?;
163                cols.push(ColumnInfo::new(name.clone(), meta.data_type.clone()));
164            }
165            Ok(cols)
166        }
167        Projection::Columns(cols) => {
168            let mut result = Vec::with_capacity(cols.len());
169            for (i, col) in cols.iter().enumerate() {
170                let name = col
171                    .alias
172                    .clone()
173                    .or_else(|| {
174                        if let alopex_sql::planner::typed_expr::TypedExprKind::ColumnRef {
175                            column,
176                            ..
177                        } = &col.expr.kind
178                        {
179                            Some(column.clone())
180                        } else {
181                            None
182                        }
183                    })
184                    .unwrap_or_else(|| format!("col_{}", i));
185                result.push(ColumnInfo::new(name, col.expr.resolved_type.clone()));
186            }
187            Ok(result)
188        }
189    }
190}
191
192impl Database {
193    /// SQL を実行する（auto-commit）。
194    ///
195    /// - DDL/DML は ReadWrite トランザクションで実行し、成功時に自動コミットする。
196    /// - SELECT は ReadOnly トランザクションで実行する。
197    /// - 複数文はすべて同一トランザクションで実行し、最後の文の結果を返す。
198    ///   文ごとの結果が必要な場合は [`Database::execute_sql_multi`] を使用する。
199    ///
200    /// # Examples
201    ///
202    /// ```
203    /// use alopex_embedded::Database;
204    /// use alopex_sql::ExecutionResult;
205    ///
206    /// let db = Database::new();
207    /// let result = db.execute_sql(
208    ///     "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);",
209    /// ).unwrap();
210    /// assert!(matches!(result, ExecutionResult::Success));
211    /// ```
212    pub fn execute_sql(&self, sql: &str) -> Result<SqlResult> {
213        Ok(self
214            .execute_sql_multi(sql)?
215            .pop()
216            .unwrap_or(alopex_sql::ExecutionResult::Success))
217    }
218
219    /// SQL を実行し、文ごとの実行結果を返す（auto-commit）。
220    ///
221    /// すべての文を同一トランザクションで実行し、成功時に自動コミットする。
222    /// いずれかの文が失敗した場合はトランザクション全体がロールバックされ、
223    /// エラーを返す。入力に文が含まれない場合は空の `Vec` を返す。
224    ///
225    /// # Examples
226    ///
227    /// ```
228    /// use alopex_embedded::Database;
229    /// use alopex_sql::ExecutionResult;
230    ///
231    /// let db = Database::new();
232    /// let results = db.execute_sql_multi(
233    ///     "CREATE TABLE users (id INTEGER PRIMARY KEY); INSERT INTO users (id) VALUES (1);",
234    /// ).unwrap();
235    /// assert_eq!(results.len(), 2);
236    /// assert!(matches!(results[0], ExecutionResult::Success));
237    /// assert!(matches!(results[1], ExecutionResult::RowsAffected(1)));
238    /// ```
239    pub fn execute_sql_multi(&self, sql: &str) -> Result<Vec<SqlResult>> {
240        let stmts = parse_sql(sql)?;
241        if stmts.is_empty() {
242            return Ok(Vec::new());
243        }
244
245        // PRAGMA controls must execute against the store itself. Running them
246        // through the external-transaction bridge would hide the store from
247        // the executor and correctly reject the operation. Keep the
248        // auto-commit public API usable for a standalone PRAGMA.
249        if stmts.len() == 1 {
250            let overlay = CatalogOverlay::new();
251            let plan = {
252                let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
253                plan_stmt(&*catalog, &overlay, &stmts[0])?
254            };
255            if matches!(stmts[0].kind, StatementKind::Pragma { .. })
256                || alopex_sql::executor::is_store_direct_plan(&plan)
257            {
258                let mut executor: Executor<_, _> =
259                    Executor::new(self.store.clone(), self.sql_catalog.clone());
260                return executor
261                    .execute(plan)
262                    .map(|result| vec![result])
263                    .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)));
264            }
265        }
266
267        let requires_write = stmts.iter().any(stmt_requires_write);
268        let mode = if requires_write {
269            TxnMode::ReadWrite
270        } else {
271            TxnMode::ReadOnly
272        };
273
274        let mut txn = self.store.begin(mode).map_err(Error::Core)?;
275        let mut overlay = CatalogOverlay::new();
276        let mut borrowed =
277            TxnBridge::<alopex_core::kv::AnyKV>::wrap_external(&mut txn, mode, &mut overlay);
278
279        let mut executor: Executor<_, _> =
280            Executor::new(self.store.clone(), self.sql_catalog.clone());
281
282        let mut results = Vec::with_capacity(stmts.len());
283        for (statement_index, stmt) in stmts.iter().enumerate() {
284            let plan = {
285                let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
286                let (_, overlay) = borrowed.split_parts();
287                plan_stmt(&*catalog, &*overlay, stmt)?
288            };
289
290            {
291                let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
292                let (_, overlay) = borrowed.split_parts();
293                let view = TxnCatalogView::new(&*catalog, &*overlay);
294                self.record_routing(&view, stmt, statement_index);
295            }
296
297            results.push(
298                executor
299                    .execute_in_txn(plan, &mut borrowed)
300                    .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?,
301            );
302        }
303
304        drop(borrowed);
305
306        // `execute_in_txn()` 成功時に HNSW flush 済み（失敗時は abandon 済み）なので、
307        // ここでは KV commit と overlay 適用のみを行う。
308        //
309        // commit_self は `txn` を消費するため、失敗時に rollback はできない。
310        txn.commit_self().map_err(Error::Core)?;
311        if mode == TxnMode::ReadWrite {
312            let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
313            catalog.apply_overlay(overlay);
314        }
315        if stmts.iter().any(stmt_changes_catalog) {
316            self.invalidate_table_info_cache();
317        }
318        if requires_write {
319            let mut cache = self.hnsw_cache.write().expect("hnsw cache lock poisoned");
320            cache.clear();
321            let mut vector_cache = self
322                .vector_cache
323                .write()
324                .expect("vector cache lock poisoned");
325            *vector_cache = None;
326        }
327        Ok(results)
328    }
329
330    /// Execute SQL with callback-based streaming for SELECT queries (FR-7).
331    ///
332    /// This method provides true streaming by keeping the transaction alive
333    /// during row iteration. The callback receives a `StreamingRows` that
334    /// yields rows on-demand from storage.
335    ///
336    /// # Type Parameters
337    ///
338    /// * `F` - Callback function that processes the streaming rows
339    /// * `R` - Return type from the callback
340    ///
341    /// # Examples
342    ///
343    /// ```
344    /// use alopex_embedded::Database;
345    ///
346    /// let db = Database::new();
347    /// db.execute_sql("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);").unwrap();
348    /// db.execute_sql("INSERT INTO users (id, name) VALUES (1, 'Alice'), (2, 'Bob');").unwrap();
349    ///
350    /// // Process rows with streaming - transaction stays alive during callback.
351    /// // Propagate `next_row` errors with `?`; swallowing them (e.g. with
352    /// // `while let Ok(Some(...))`) silently truncates the result set.
353    /// let result = db.execute_sql_with_rows("SELECT * FROM users;", |mut rows| {
354    ///     let mut names = Vec::new();
355    ///     while let Some(row) = rows.next_row()? {
356    ///         if let Some(alopex_sql::storage::SqlValue::Text(name)) = row.get(1) {
357    ///             names.push(name.clone());
358    ///         }
359    ///     }
360    ///     Ok(names)
361    /// }).unwrap();
362    /// ```
363    pub fn execute_sql_with_rows<F, R>(&self, sql: &str, f: F) -> Result<StreamingQueryResult<R>>
364    where
365        F: FnOnce(StreamingRows<'_>) -> Result<R>,
366    {
367        let stmts = parse_sql(sql)?;
368        if stmts.is_empty() {
369            return Ok(StreamingQueryResult::Success);
370        }
371
372        // For streaming SELECT, use the new pipeline
373        if stmts.len() == 1 && matches!(stmts[0].kind, StatementKind::Select(_)) {
374            let stmt = &stmts[0];
375            let mode = TxnMode::ReadOnly;
376
377            let mut txn = self.store.begin(mode).map_err(Error::Core)?;
378            let mut overlay = CatalogOverlay::new();
379            let mut borrowed =
380                TxnBridge::<alopex_core::kv::AnyKV>::wrap_external(&mut txn, mode, &mut overlay);
381
382            let plan = {
383                let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
384                let (_, overlay_ref) = borrowed.split_parts();
385                plan_stmt(&*catalog, overlay_ref, stmt)?
386            };
387
388            if alopex_sql::executor::is_store_direct_plan(&plan) {
389                let mut executor: Executor<_, _> =
390                    Executor::new(self.store.clone(), self.sql_catalog.clone());
391                let result = executor
392                    .execute(plan)
393                    .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
394                let ExecutionResult::Query(query) = result else {
395                    return Err(Error::Sql(alopex_sql::SqlError::Execution {
396                        message: "store-direct system function did not return rows".into(),
397                        code: "ALOPEX-E022",
398                    }));
399                };
400                let column_names: Vec<String> = query
401                    .columns
402                    .iter()
403                    .map(|column| column.name.clone())
404                    .collect();
405                let schema: Vec<alopex_sql::catalog::ColumnMetadata> = query
406                    .columns
407                    .iter()
408                    .map(|column| {
409                        alopex_sql::catalog::ColumnMetadata::new(
410                            &column.name,
411                            column.data_type.clone(),
412                        )
413                    })
414                    .collect();
415                let rows: Vec<Row> = query
416                    .rows
417                    .into_iter()
418                    .enumerate()
419                    .map(|(index, values)| Row::new(index as u64, values))
420                    .collect();
421                let iter = VecIterator::new(rows, schema.clone());
422                let streaming_rows = StreamingRows {
423                    columns: query.columns,
424                    iter: Box::new(iter),
425                    projection: Projection::All(column_names),
426                    schema,
427                };
428                let result = f(streaming_rows)?;
429                return Ok(StreamingQueryResult::QueryProcessed(result));
430            }
431
432            let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
433            let (mut sql_txn, overlay_ref) = borrowed.split_parts();
434            let view = TxnCatalogView::new(&*catalog, overlay_ref);
435
436            // Build streaming pipeline - iterator lifetime tied to sql_txn
437            let (iter, projection, schema) = build_streaming_pipeline(&mut sql_txn, &view, plan)
438                .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
439
440            // Build column info from projection and schema
441            let columns = build_column_info(&projection, &schema)?;
442
443            // Create StreamingRows and pass to callback
444            let streaming_rows = StreamingRows {
445                columns,
446                iter,
447                projection,
448                schema,
449            };
450
451            // Execute callback with streaming rows
452            let result = f(streaming_rows)?;
453
454            // Clean up and commit after callback completes
455            drop(catalog);
456            drop(borrowed);
457            txn.commit_self().map_err(Error::Core)?;
458
459            return Ok(StreamingQueryResult::QueryProcessed(result));
460        }
461
462        // Fall back to standard execution for non-SELECT or multi-statement
463        let exec_result = self.execute_sql(sql)?;
464        match exec_result {
465            alopex_sql::ExecutionResult::Success => Ok(StreamingQueryResult::Success),
466            alopex_sql::ExecutionResult::RowsAffected(n) => {
467                Ok(StreamingQueryResult::RowsAffected(n))
468            }
469            alopex_sql::ExecutionResult::Query(_qr) => {
470                // For non-streaming path, we can't provide true streaming
471                // Return an error indicating streaming is not available
472                Err(Error::Sql(alopex_sql::SqlError::Execution {
473                    message: "Streaming not available for multi-statement or complex queries"
474                        .into(),
475                    code: "ALOPEX-E021",
476                }))
477            }
478        }
479    }
480
481    /// Execute SQL and return a streaming result for SELECT queries (FR-7).
482    ///
483    /// This method returns a `SqlStreamingResult` that contains an iterator
484    /// for query results, enabling true streaming output without materializing
485    /// all rows upfront.
486    ///
487    /// # Note
488    ///
489    /// Only single SELECT statements are supported for streaming. Multi-statement
490    /// SQL or non-SELECT statements fall back to the standard execution path.
491    ///
492    /// # Examples
493    ///
494    /// ```
495    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
496    /// use alopex_embedded::{Database, SqlStreamingResult};
497    ///
498    /// let db = Database::new();
499    /// db.execute_sql("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);")?;
500    /// db.execute_sql("INSERT INTO users (id, name) VALUES (1, 'Alice');")?;
501    ///
502    /// // Propagate `next_row` errors with `?`; swallowing them (e.g. with
503    /// // `while let Ok(Some(...))`) silently truncates the result set.
504    /// let result = db.execute_sql_streaming("SELECT * FROM users;")?;
505    /// if let SqlStreamingResult::Query(mut iter) = result {
506    ///     while let Some(row) = iter.next_row()? {
507    ///         println!("{:?}", row);
508    ///     }
509    /// }
510    /// # Ok(())
511    /// # }
512    /// ```
513    pub fn execute_sql_streaming(&self, sql: &str) -> Result<SqlStreamingResult> {
514        let stmts = parse_sql(sql)?;
515        if stmts.is_empty() {
516            return Ok(SqlStreamingResult::Success);
517        }
518
519        // For streaming, only support single SELECT statement
520        if stmts.len() == 1 && matches!(stmts[0].kind, StatementKind::Select(_)) {
521            let stmt = &stmts[0];
522            let mode = TxnMode::ReadOnly;
523
524            let mut txn = self.store.begin(mode).map_err(Error::Core)?;
525            let mut overlay = CatalogOverlay::new();
526            let mut borrowed =
527                TxnBridge::<alopex_core::kv::AnyKV>::wrap_external(&mut txn, mode, &mut overlay);
528
529            let plan = {
530                let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
531                let (_, overlay) = borrowed.split_parts();
532                plan_stmt(&*catalog, &*overlay, stmt)?
533            };
534
535            let (mut sql_txn, _overlay) = borrowed.split_parts();
536
537            let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
538            let view = TxnCatalogView::new(&*catalog, _overlay);
539            let iter = execute_query_streaming(&mut sql_txn, &view, plan)
540                .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
541
542            drop(catalog);
543            drop(borrowed);
544
545            txn.commit_self().map_err(Error::Core)?;
546
547            return Ok(SqlStreamingResult::Query(iter));
548        }
549
550        // Fall back to standard execution for non-streaming cases
551        let result = self.execute_sql(sql)?;
552        match result {
553            alopex_sql::ExecutionResult::Success => Ok(SqlStreamingResult::Success),
554            alopex_sql::ExecutionResult::RowsAffected(n) => Ok(SqlStreamingResult::RowsAffected(n)),
555            alopex_sql::ExecutionResult::Query(qr) => {
556                // Convert materialized result to streaming iterator
557                use alopex_sql::executor::query::iterator::VecIterator;
558                use alopex_sql::executor::Row;
559                use alopex_sql::planner::typed_expr::Projection;
560
561                let column_names: Vec<String> = qr.columns.iter().map(|c| c.name.clone()).collect();
562                let schema: Vec<alopex_sql::catalog::ColumnMetadata> = qr
563                    .columns
564                    .iter()
565                    .map(|c| alopex_sql::catalog::ColumnMetadata::new(&c.name, c.data_type.clone()))
566                    .collect();
567                let rows: Vec<Row> = qr
568                    .rows
569                    .into_iter()
570                    .enumerate()
571                    .map(|(i, values)| Row::new(i as u64, values))
572                    .collect();
573                let iter = VecIterator::new(rows, schema.clone());
574                let query_iter =
575                    QueryRowIterator::new(Box::new(iter), Projection::All(column_names), schema);
576                Ok(SqlStreamingResult::Query(query_iter))
577            }
578        }
579    }
580}
581
582impl<'a> Transaction<'a> {
583    /// SQL を実行する（外部トランザクション利用）。
584    ///
585    /// 同一トランザクション内の複数回呼び出しでカタログ変更が見えるよう、`CatalogOverlay` は
586    /// `Transaction` が所有して保持する。
587    ///
588    /// # Examples
589    ///
590    /// ```
591    /// use alopex_embedded::{Database, TxnMode};
592    ///
593    /// let db = Database::new();
594    /// let mut txn = db.begin(TxnMode::ReadWrite).unwrap();
595    /// txn.execute_sql("CREATE TABLE t (id INTEGER PRIMARY KEY);").unwrap();
596    /// txn.execute_sql("INSERT INTO t (id) VALUES (1);").unwrap();
597    /// txn.commit().unwrap();
598    /// ```
599    pub fn execute_sql(&mut self, sql: &str) -> Result<SqlResult> {
600        let stmts = parse_sql(sql)?;
601        if stmts.is_empty() {
602            return Ok(alopex_sql::ExecutionResult::Success);
603        }
604
605        if stmts.iter().any(stmt_requires_write) {
606            let mut cache = self
607                .db
608                .hnsw_cache
609                .write()
610                .expect("hnsw cache lock poisoned");
611            cache.clear();
612            let mut vector_cache = self
613                .db
614                .vector_cache
615                .write()
616                .expect("vector cache lock poisoned");
617            *vector_cache = None;
618        }
619
620        let store = self.db.store.clone();
621        let sql_catalog = self.db.sql_catalog.clone();
622
623        let txn = self.inner.as_mut().ok_or(Error::TxnCompleted)?;
624        let mode = txn.mode();
625
626        let mut borrowed =
627            TxnBridge::<alopex_core::kv::AnyKV>::wrap_external(txn, mode, &mut self.overlay);
628        let mut executor: Executor<_, _> = Executor::new(store, sql_catalog.clone());
629
630        let mut last = alopex_sql::ExecutionResult::Success;
631        for (statement_index, stmt) in stmts.iter().enumerate() {
632            let plan = {
633                let catalog = sql_catalog.read().expect("catalog lock poisoned");
634                let (_, overlay) = borrowed.split_parts();
635                plan_stmt(&*catalog, &*overlay, stmt)?
636            };
637
638            {
639                let catalog = sql_catalog.read().expect("catalog lock poisoned");
640                let (_, overlay) = borrowed.split_parts();
641                let view = TxnCatalogView::new(&*catalog, &*overlay);
642                self.db.record_routing(&view, stmt, statement_index);
643            }
644
645            last = executor
646                .execute_in_txn(plan, &mut borrowed)
647                .map_err(|e| Error::Sql(alopex_sql::SqlError::from(e)))?;
648        }
649
650        if stmts.iter().any(stmt_changes_catalog) {
651            self.catalog_modified = true;
652        }
653        Ok(last)
654    }
655}
alopex_embedded/sql_api.rs

alopex_embedded/
sql_api.rs
