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

alopex_embedded/
sql_api.rs
