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

alopex_embedded/
sql_api.rs
