teaql_provider_sqlite/
lib.rs

1use std::collections::{BTreeMap, BTreeSet, HashMap};
2use std::future::Future;
3use std::pin::Pin;
4use std::str::FromStr;
5use std::sync::{Arc, Mutex, MutexGuard};
6
7use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, TimeZone};
8use rusqlite::types::{Value as SqliteValue, ValueRef};
9use rusqlite::{
10    Connection, OptionalExtension, Row, functions::FunctionFlags, params, params_from_iter,
11};
12use rust_decimal::Decimal;
13use teaql_core::{
14    CompactRow, DataType, EntityDescriptor, Expr, InsertCommand, PropertyDescriptor, SelectQuery,
15    UpdateCommand, Value,
16};
17use teaql_runtime::{
18    GraphNode, InternalIdGenerator, RawAuditEvent, RuntimeError, SchemaProvider, UserContext,
19    canonical_id_space_entity,
20};
21use teaql_sql::{
22    CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, SqlTransport,
23    quote_identifier_if_needed,
24};
25
26pub const DEFAULT_ID_SPACE_TABLE: &str = "teaql_id_space";
27pub const DEFAULT_PREPARED_STATEMENT_CACHE_CAPACITY: usize = 64;
28pub const DEFAULT_COLUMN_LAYOUT_CACHE_CAPACITY: usize = 64;
29
30#[derive(Debug, Default, Clone, Copy)]
31pub struct SqliteDialect;
32
33impl SqlDialect for SqliteDialect {
34    fn kind(&self) -> DatabaseKind {
35        DatabaseKind::Sqlite
36    }
37
38    fn quote_ident(&self, ident: &str) -> String {
39        quote_ident(ident)
40    }
41
42    fn placeholder(&self, _index: usize) -> String {
43        "?".to_owned()
44    }
45
46    fn prefers_small_parent_relation_probes(&self) -> bool {
47        true
48    }
49
50    fn schema_type_sql(
51        &self,
52        data_type: DataType,
53        property: &PropertyDescriptor,
54    ) -> Result<&'static str, SqlCompileError> {
55        match data_type {
56            DataType::Bool => Ok("BOOLEAN"),
57            DataType::I64 | DataType::U64 if property.is_id => Ok("INTEGER"),
58            DataType::I64 | DataType::U64 => Ok("INTEGER"),
59            DataType::F64 => Ok("REAL"),
60            DataType::Decimal => Ok("NUMERIC"),
61            DataType::Text => Ok("VARCHAR(255)"),
62            DataType::LargeText => Ok("TEXT"),
63            DataType::Json => Ok("JSON"),
64            DataType::Date => Ok("DATE"),
65            DataType::Timestamp => Ok("TIMESTAMP"),
66        }
67    }
68
69    fn compile_add_column(
70        &self,
71        entity: &EntityDescriptor,
72        property: &PropertyDescriptor,
73    ) -> Result<String, SqlCompileError> {
74        // SQLite does not support adding NOT NULL columns without a DEFAULT.
75        // Since TeaQL enforces nullability at the application layer, we can safely
76        // strip the NOT NULL constraint when adding columns to existing tables.
77        let def = self.column_definition_sql(property)?;
78        let def_without_not_null = def.replace(" NOT NULL", "");
79
80        Ok(format!(
81            "ALTER TABLE {} ADD COLUMN {}",
82            self.quote_ident(&entity.table_name),
83            def_without_not_null
84        ))
85    }
86}
87
88#[derive(Debug)]
89pub enum MutationExecutorError {
90    Sqlite(rusqlite::Error),
91    SqlCompile(SqlCompileError),
92    UnsupportedValue(&'static str),
93    UnsupportedColumnType(String),
94    Bind(String),
95    Lock(String),
96}
97
98impl std::fmt::Display for MutationExecutorError {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        match self {
101            Self::Sqlite(err) => err.fmt(f),
102            Self::SqlCompile(err) => err.fmt(f),
103            Self::UnsupportedValue(kind) => {
104                write!(
105                    f,
106                    "unsupported rusqlite bind value for mutation executor: {kind}"
107                )
108            }
109            Self::UnsupportedColumnType(kind) => {
110                write!(
111                    f,
112                    "unsupported rusqlite column type for record decoding: {kind}"
113                )
114            }
115            Self::Bind(message) => write!(f, "rusqlite bind error: {message}"),
116            Self::Lock(message) => write!(f, "rusqlite connection lock error: {message}"),
117        }
118    }
119}
120
121impl std::error::Error for MutationExecutorError {}
122
123impl From<rusqlite::Error> for MutationExecutorError {
124    fn from(value: rusqlite::Error) -> Self {
125        Self::Sqlite(value)
126    }
127}
128
129impl From<SqlCompileError> for MutationExecutorError {
130    fn from(value: SqlCompileError) -> Self {
131        Self::SqlCompile(value)
132    }
133}
134
135#[derive(Clone)]
136pub struct SqliteMutationExecutor {
137    connection: Arc<Mutex<Connection>>,
138    column_layout_cache: Arc<Mutex<HashMap<String, Arc<ColumnLayout>>>>,
139}
140
141impl SqliteMutationExecutor {
142    pub fn new(connection: Arc<Mutex<Connection>>) -> Self {
143        if let Ok(connection) = connection.lock() {
144            connection
145                .set_prepared_statement_cache_capacity(DEFAULT_PREPARED_STATEMENT_CACHE_CAPACITY);
146        }
147        Self {
148            connection,
149            column_layout_cache: Arc::new(Mutex::new(HashMap::new())),
150        }
151    }
152
153    pub fn from_connection(connection: Connection) -> Self {
154        Self::new(Arc::new(Mutex::new(connection)))
155    }
156
157    pub fn connection(&self) -> Arc<Mutex<Connection>> {
158        Arc::clone(&self.connection)
159    }
160
161    pub fn ensure_schema(
162        &self,
163        dialect: &SqliteDialect,
164        entities: &[&EntityDescriptor],
165    ) -> Result<(), MutationExecutorError> {
166        self.ensure_soundex_function()?;
167        self.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
168
169        for entity in entities {
170            if !self.table_exists(&entity.table_name)? {
171                let sql = dialect.compile_create_table(entity)?;
172                self.lock()?.execute(&sql, [])?;
173                continue;
174            }
175
176            let existing_columns = self.table_columns(&entity.table_name)?;
177            for property in &entity.properties {
178                let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
179                if existing_columns.contains(&bare_column) {
180                    continue;
181                }
182                let sql = dialect.compile_add_column(entity, property)?;
183                self.lock()?.execute(&sql, [])?;
184            }
185
186            for sql in dialect.schema_indexes_sqls(entity)? {
187                self.lock()?.execute(&sql, [])?;
188            }
189        }
190        self.clear_query_caches();
191        Ok(())
192    }
193
194    fn ensure_soundex_function(&self) -> Result<(), MutationExecutorError> {
195        self.lock()?.create_scalar_function(
196            "soundex",
197            1,
198            FunctionFlags::SQLITE_UTF8 | FunctionFlags::SQLITE_DETERMINISTIC,
199            |ctx| {
200                let input = ctx.get_raw(0).as_str().ok();
201                Ok(sqlite_compatible_soundex(input))
202            },
203        )?;
204        Ok(())
205    }
206
207    fn clear_query_caches(&self) {
208        if let Ok(connection) = self.connection.lock() {
209            connection.flush_prepared_statement_cache();
210        }
211        if let Ok(mut cache) = self.column_layout_cache.lock() {
212            cache.clear();
213        }
214    }
215
216    pub fn ensure_id_space_table(&self, table_name: &str) -> Result<(), MutationExecutorError> {
217        let sql = format!(
218            "CREATE TABLE IF NOT EXISTS {} (type_name VARCHAR(100) PRIMARY KEY, current_level BIGINT NOT NULL)",
219            quote_ident(table_name)
220        );
221        self.lock()?.execute(&sql, [])?;
222        Ok(())
223    }
224
225    pub fn begin_transaction(&self) -> Result<(), MutationExecutorError> {
226        self.lock()?.execute("BEGIN IMMEDIATE", [])?;
227        Ok(())
228    }
229
230    pub fn commit_transaction(&self) -> Result<(), MutationExecutorError> {
231        self.lock()?.execute("COMMIT", [])?;
232        Ok(())
233    }
234
235    pub fn rollback_transaction(&self) -> Result<(), MutationExecutorError> {
236        self.lock()?.execute("ROLLBACK", [])?;
237        Ok(())
238    }
239
240    pub fn execute(&self, query: &CompiledQuery) -> Result<u64, MutationExecutorError> {
241        let params = bind_values(&query.params)?;
242        let rows = self
243            .lock()?
244            .execute(&query.sql_with_comment(), params_from_iter(params.iter()))?;
245        Ok(rows as u64)
246    }
247
248    pub fn fetch_all_compact(
249        &self,
250        query: &CompiledQuery,
251    ) -> Result<Vec<CompactRow>, MutationExecutorError> {
252        let params = bind_values(&query.params)?;
253        let connection = self.lock()?;
254        let sql = query.sql_with_comment();
255        let mut statement = connection.prepare_cached(&sql)?;
256        let layout = cached_column_layout(&self.column_layout_cache, &query.sql, &statement);
257        let mut rows = statement.query(params_from_iter(params.iter()))?;
258        let mut result = Vec::new();
259        while let Some(row) = rows.next()? {
260            result.push(CompactRow::new(
261                layout.names.clone(),
262                decode_sqlite_values(row, &layout.columns)?,
263            ));
264        }
265        Ok(result)
266    }
267
268    /// Fetch rows in streaming mode (chunked).
269    /// Returns a Vec of StreamChunk, each containing up to `chunk_size` rows.
270    pub fn fetch_stream(
271        &self,
272        query: &CompiledQuery,
273        chunk_size: usize,
274    ) -> Result<Vec<teaql_data_service::StreamChunk>, MutationExecutorError> {
275        let params = bind_values(&query.params)?;
276        let connection = self.lock()?;
277        let sql = query.sql_with_comment();
278        let mut statement = connection.prepare_cached(&sql)?;
279        let layout = cached_column_layout(&self.column_layout_cache, &query.sql, &statement);
280        let mut rows = statement.query(params_from_iter(params.iter()))?;
281
282        let mut chunks = Vec::new();
283        let mut current_chunk = Vec::new();
284        let mut chunk_index = 0;
285
286        while let Some(row) = rows.next()? {
287            current_chunk.push(CompactRow::new(
288                layout.names.clone(),
289                decode_sqlite_values(row, &layout.columns)?,
290            ));
291            if current_chunk.len() >= chunk_size {
292                chunks.push(teaql_data_service::StreamChunk {
293                    rows: current_chunk,
294                    chunk_index,
295                    is_last: false,
296                });
297                current_chunk = Vec::new();
298                chunk_index += 1;
299            }
300        }
301
302        // Push the final chunk (may be empty if exactly aligned)
303        chunks.push(teaql_data_service::StreamChunk {
304            rows: current_chunk,
305            chunk_index,
306            is_last: true,
307        });
308
309        Ok(chunks)
310    }
311
312    pub fn table_exists(&self, table_name: &str) -> Result<bool, MutationExecutorError> {
313        let exists: i64 = self.lock()?.query_row(
314            "SELECT COUNT(1) FROM sqlite_master WHERE type = 'table' AND name = ?",
315            [table_name],
316            |row| row.get(0),
317        )?;
318        Ok(exists > 0)
319    }
320
321    pub fn table_columns(
322        &self,
323        table_name: &str,
324    ) -> Result<BTreeSet<String>, MutationExecutorError> {
325        let pragma_sql = format!("PRAGMA table_info({})", quote_ident(table_name));
326        let connection = self.lock()?;
327        let mut statement = connection.prepare(&pragma_sql)?;
328        let rows = statement.query_map([], |row| row.get::<_, String>("name"))?;
329        let mut columns = BTreeSet::new();
330        for row in rows {
331            columns.insert(row?.to_lowercase());
332        }
333        Ok(columns)
334    }
335
336    fn lock(&self) -> Result<MutexGuard<'_, Connection>, MutationExecutorError> {
337        self.connection
338            .lock()
339            .map_err(|err| MutationExecutorError::Lock(err.to_string()))
340    }
341}
342
343fn sqlite_compatible_soundex(input: Option<&str>) -> String {
344    fn code(byte: u8) -> u8 {
345        match byte.to_ascii_uppercase() {
346            b'B' | b'F' | b'P' | b'V' => 1,
347            b'C' | b'G' | b'J' | b'K' | b'Q' | b'S' | b'X' | b'Z' => 2,
348            b'D' | b'T' => 3,
349            b'L' => 4,
350            b'M' | b'N' => 5,
351            b'R' => 6,
352            _ => 0,
353        }
354    }
355    let Some(input) = input else {
356        return "?000".to_owned();
357    };
358    let Some((first_index, first)) = input
359        .bytes()
360        .enumerate()
361        .find(|(_, byte)| byte.is_ascii_alphabetic())
362    else {
363        return "?000".to_owned();
364    };
365    let mut result = String::with_capacity(4);
366    result.push(char::from(first.to_ascii_uppercase()));
367    let mut previous = code(first);
368    for byte in input.bytes().skip(first_index + 1) {
369        if !byte.is_ascii_alphabetic() {
370            continue;
371        }
372        let current = code(byte);
373        if current != 0 && current != previous {
374            result.push(char::from(b'0' + current));
375            if result.len() == 4 {
376                break;
377            }
378        }
379        previous = current;
380    }
381    while result.len() < 4 {
382        result.push('0');
383    }
384    result
385}
386
387impl teaql_data_service::DataServiceExecutor for SqliteMutationExecutor {
388    type Error = MutationExecutorError;
389
390    fn capabilities(&self) -> teaql_data_service::DataServiceCapabilities {
391        teaql_data_service::DataServiceCapabilities {
392            query: true,
393            mutation: true,
394            transaction: true,
395            schema: true,
396            id_generation: true,
397            ..Default::default()
398        }
399    }
400}
401
402impl SqlTransport for SqliteMutationExecutor {
403    type Error = MutationExecutorError;
404
405    async fn fetch_all_compact_sql(
406        &self,
407        query: &CompiledQuery,
408    ) -> Result<Vec<CompactRow>, Self::Error> {
409        SqliteMutationExecutor::fetch_all_compact(self, query)
410    }
411
412    async fn fetch_repeated_compact_sql(
413        &self,
414        template: &CompiledQuery,
415        param_index: usize,
416        values: &[Value],
417    ) -> Result<Vec<CompactRow>, Self::Error> {
418        let connection = self.lock()?;
419        let sql = template.sql_with_comment();
420        let mut statement = connection.prepare_cached(&sql)?;
421        let layout = cached_column_layout(&self.column_layout_cache, &template.sql, &statement);
422        let mut result = Vec::new();
423        let mut query_params = template.params.clone();
424        for value in values {
425            query_params[param_index] = value.clone();
426            let params = bind_values(&query_params)?;
427            let mut rows = statement.query(params_from_iter(params.iter()))?;
428            while let Some(row) = rows.next()? {
429                result.push(CompactRow::new(
430                    layout.names.clone(),
431                    decode_sqlite_values(row, &layout.columns)?,
432                ));
433            }
434        }
435        Ok(result)
436    }
437
438    async fn execute_sql(&self, query: &CompiledQuery) -> Result<u64, Self::Error> {
439        SqliteMutationExecutor::execute(self, query)
440    }
441}
442
443impl teaql_sql::StreamingSqlTransport for SqliteMutationExecutor {
444    // A rusqlite statement and its rows borrow the guarded connection for the
445    // lifetime of the stream. This stream is intentionally local/non-Send, so
446    // retaining the synchronous guard across yields is required and safe.
447    #[allow(clippy::await_holding_lock)]
448    fn stream_sql(
449        &self,
450        query: CompiledQuery,
451        chunk_size: usize,
452    ) -> teaql_data_service::QueryStream<'_, Self::Error> {
453        let connection = self.connection.clone();
454        let column_layout_cache = self.column_layout_cache.clone();
455        Box::pin(async_stream::try_stream! {
456            let params = bind_values(&query.params)?;
457            let guard = connection.lock().map_err(|err| MutationExecutorError::Lock(err.to_string()))?;
458            let sql = query.sql_with_comment();
459            let mut statement = guard.prepare_cached(&sql)?;
460            let layout = cached_column_layout(&column_layout_cache, &query.sql, &statement);
461            let mut rows = statement.query(params_from_iter(params.iter()))?;
462            let mut chunk = Vec::with_capacity(chunk_size); let mut index = 0;
463            while let Some(row) = rows.next()? {
464                chunk.push(CompactRow::new(layout.names.clone(), decode_sqlite_values(row, &layout.columns)?));
465                if chunk.len() == chunk_size { yield teaql_data_service::StreamChunk { rows: std::mem::take(&mut chunk), chunk_index: index, is_last: false }; index += 1; }
466            }
467            if !chunk.is_empty() { yield teaql_data_service::StreamChunk { rows: chunk, chunk_index: index, is_last: true }; }
468        })
469    }
470}
471
472impl teaql_data_service::StreamQueryExecutor for SqliteMutationExecutor {
473    fn query_stream(
474        &self,
475        request: teaql_data_service::QueryRequest,
476        chunk_size: usize,
477    ) -> teaql_data_service::QueryStream<'_, Self::Error> {
478        let dialect = SqliteDialect;
479        // Use a dummy entity descriptor for compilation
480        let entity_desc = teaql_core::EntityDescriptor::new(&request.query.entity);
481        match dialect.compile_select(&entity_desc, &request.query) {
482            Ok(compiled) => {
483                teaql_sql::StreamingSqlTransport::stream_sql(self, compiled, chunk_size)
484            }
485            Err(error) => Box::pin(futures_util::stream::once(async {
486                Err(MutationExecutorError::SqlCompile(error))
487            })),
488        }
489    }
490}
491
492impl teaql_sql::SqlTransaction for SqliteMutationExecutor {
493    type Error = MutationExecutorError;
494
495    async fn commit_sql(self) -> Result<(), Self::Error> {
496        self.commit_transaction()
497    }
498
499    async fn rollback_sql(self) -> Result<(), Self::Error> {
500        self.rollback_transaction()
501    }
502}
503
504impl teaql_sql::SqlTransactionTransport for SqliteMutationExecutor {
505    type Tx<'a>
506        = Self
507    where
508        Self: 'a;
509
510    async fn begin_sql(&self) -> Result<Self::Tx<'_>, Self::Error> {
511        self.begin_transaction()?;
512        Ok(self.clone())
513    }
514}
515
516fn initial_graph_row_sqlite(
517    executor: &SqliteMutationExecutor,
518    dialect: &SqliteDialect,
519    entity: &EntityDescriptor,
520    graph: &GraphNode,
521) -> Result<Option<teaql_core::CompactRow>, MutationExecutorError> {
522    let Some(id) = graph.values.get("id") else {
523        return Ok(None);
524    };
525    let mut select = SelectQuery::new(&graph.entity)
526        .filter(Expr::eq("id", id.clone()))
527        .limit(1);
528    for field in graph.values.keys() {
529        select = select.project(field);
530    }
531    if let Some(version) = entity
532        .version_property()
533        .filter(|version| !graph.values.contains_key(&version.name))
534    {
535        select = select.project(&version.name);
536    }
537    let query = dialect.compile_select(entity, &select)?;
538    Ok(executor.fetch_all_compact(&query)?.into_iter().next())
539}
540
541fn compile_initial_graph_insert(
542    dialect: &impl SqlDialect,
543    entity: &EntityDescriptor,
544    graph: &GraphNode,
545) -> Result<CompiledQuery, MutationExecutorError> {
546    let mut command = InsertCommand::new(&graph.entity);
547    for (field, value) in &graph.values {
548        command = command.value(field.clone(), value.clone());
549    }
550    dialect.compile_insert(entity, &command).map_err(Into::into)
551}
552
553fn compile_initial_graph_update(
554    dialect: &impl SqlDialect,
555    entity: &EntityDescriptor,
556    graph: &GraphNode,
557    current: &teaql_core::CompactRow,
558) -> Result<Option<CompiledQuery>, MutationExecutorError> {
559    let Some(id) = graph.values.get("id") else {
560        return Ok(None);
561    };
562    let mut command = UpdateCommand::new(&graph.entity, id.clone());
563    for (field, value) in &graph.values {
564        if field != "id"
565            && field != "version"
566            && !bootstrap_values_equal(current.get(field), Some(value))
567        {
568            command = command.value(field.clone(), value.clone());
569        }
570    }
571    if command.values.is_empty() {
572        return Ok(None);
573    }
574    if let Some(version) = entity
575        .version_property()
576        .and_then(|property| current.get(&property.name))
577        .and_then(Value::try_i64)
578    {
579        command = command.expected_version(version);
580    }
581    match dialect.compile_update(entity, &command) {
582        Ok(query) => Ok(Some(query)),
583        Err(SqlCompileError::EmptyMutation(_)) => Ok(None),
584        Err(err) => Err(err.into()),
585    }
586}
587
588fn bootstrap_values_equal(left: Option<&Value>, right: Option<&Value>) -> bool {
589    let (Some(left), Some(right)) = (left, right) else {
590        return left.is_none() && right.is_none();
591    };
592    if left == right {
593        return true;
594    }
595    matches!((left.try_decimal(), right.try_decimal()), (Some(a), Some(b)) if a == b)
596}
597
598pub(crate) fn ensure_sqlite_physical_schema_for(
599    context: &UserContext,
600) -> Result<(), MutationExecutorError> {
601    let dialect = context.get_resource::<SqliteDialect>().ok_or_else(|| {
602        MutationExecutorError::Bind("missing typed resource: SqliteDialect".to_owned())
603    })?;
604    let executor = context
605        .get_resource::<SqliteMutationExecutor>()
606        .ok_or_else(|| {
607            MutationExecutorError::Bind("missing typed resource: SqliteMutationExecutor".to_owned())
608        })?;
609
610    let entities = context.all_entities();
611
612    // Ensure id space table exists
613    executor.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
614
615    // Process each entity table individually with granular events
616    for entity in &entities {
617        let field_count = entity.properties.len();
618        if !executor.table_exists(&entity.table_name)? {
619            // New table: create it
620            let sql = dialect.compile_create_table(entity)?;
621            executor.lock()?.execute(&sql, [])?;
622            let _ = context.send_event(RawAuditEvent::schema_created(
623                &entity.name,
624                &entity.table_name,
625                field_count,
626            ));
627            continue;
628        }
629        // Existing table: check for missing columns
630        let existing_columns = executor.table_columns(&entity.table_name)?;
631        let mut fields_added = 0;
632        for property in &entity.properties {
633            let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
634            if existing_columns.contains(&bare_column) {
635                continue;
636            }
637            let sql = dialect.compile_add_column(entity, property)?;
638            executor.lock()?.execute(&sql, [])?;
639            let _ = context.send_event(RawAuditEvent::field_added(
640                &entity.name,
641                &entity.table_name,
642                &property.column_name,
643            ));
644            fields_added += 1;
645        }
646        let _ = context.send_event(RawAuditEvent::schema_verified(
647            &entity.name,
648            &entity.table_name,
649            field_count,
650        ));
651        let _ = fields_added; // used above for FieldAdded events
652    }
653
654    executor.clear_query_caches();
655    Ok(())
656}
657
658pub(crate) fn ensure_sqlite_schema_for(context: &UserContext) -> Result<(), MutationExecutorError> {
659    ensure_sqlite_physical_schema_for(context)?;
660    if !context.initial_graphs().is_empty() || !context.root_graphs().is_empty() {
661        return Err(MutationExecutorError::Bind(
662            "generated root/constant bootstrap must use the typed RuntimeModule callback"
663                .to_owned(),
664        ));
665    }
666    Ok(())
667}
668
669#[allow(dead_code)]
670fn ensure_legacy_sqlite_bootstrap_for(context: &UserContext) -> Result<(), MutationExecutorError> {
671    ensure_sqlite_physical_schema_for(context)?;
672    let dialect = context.get_resource::<SqliteDialect>().ok_or_else(|| {
673        MutationExecutorError::Bind("missing typed resource: SqliteDialect".to_owned())
674    })?;
675    let executor = context
676        .get_resource::<SqliteMutationExecutor>()
677        .ok_or_else(|| {
678            MutationExecutorError::Bind("missing typed resource: SqliteMutationExecutor".to_owned())
679        })?;
680
681    // Constant graphs are reconciled so model changes are propagated.
682    let id_generator = SqliteIdSpaceGenerator::from_executor(executor.clone());
683    let mut seed_counts: BTreeMap<String, (usize, usize)> = BTreeMap::new(); // (inserted, updated)
684    for graph in context.initial_graphs() {
685        let entity = context.entity(&graph.entity).ok_or_else(|| {
686            MutationExecutorError::Bind(format!("missing entity: {}", graph.entity))
687        })?;
688        let counts = seed_counts.entry(graph.entity.clone()).or_insert((0, 0));
689        if let Some(current) = initial_graph_row_sqlite(executor, dialect, entity, graph)? {
690            if let Some(query) = compile_initial_graph_update(dialect, entity, graph, &current)? {
691                executor.execute(&query)?;
692                counts.1 += 1;
693            }
694            if let Some(id) = graph.values.get("id").and_then(Value::try_u64) {
695                id_generator.ensure_floor(&graph.entity, id)?;
696            }
697            continue;
698        }
699        let query = compile_initial_graph_insert(dialect, entity, graph)?;
700        executor.execute(&query)?;
701        counts.0 += 1; // inserted
702        if let Some(id) = graph.values.get("id").and_then(Value::try_u64) {
703            id_generator.ensure_floor(&graph.entity, id)?;
704        }
705    }
706
707    // Roots are create-if-absent. Once present, application-owned values win.
708    for graph in context.root_graphs() {
709        let entity = context.entity(&graph.entity).ok_or_else(|| {
710            MutationExecutorError::Bind(format!("missing entity: {}", graph.entity))
711        })?;
712        if initial_graph_row_sqlite(executor, dialect, entity, graph)?.is_some() {
713            if let Some(id) = graph.values.get("id").and_then(Value::try_u64) {
714                id_generator.ensure_floor(&graph.entity, id)?;
715            }
716            continue;
717        }
718        let query = compile_initial_graph_insert(dialect, entity, graph)?;
719        executor.execute(&query)?;
720        seed_counts.entry(graph.entity.clone()).or_insert((0, 0)).0 += 1;
721        if let Some(id) = graph.values.get("id").and_then(Value::try_u64) {
722            id_generator.ensure_floor(&graph.entity, id)?;
723        }
724    }
725
726    // Fire DataSeeded events per entity type
727    for (entity_name, (inserted, updated)) in &seed_counts {
728        let entity = context.entity(entity_name).ok_or_else(|| {
729            MutationExecutorError::Bind(format!("missing entity: {}", entity_name))
730        })?;
731        let _ = context.send_event(RawAuditEvent::data_seeded(
732            entity_name,
733            &entity.table_name,
734            *inserted,
735            *updated,
736        ));
737    }
738
739    executor.clear_query_caches();
740    Ok(())
741}
742
743#[derive(Debug, Default, Clone, Copy)]
744pub struct SqliteSchemaProvider;
745
746impl SchemaProvider for SqliteSchemaProvider {
747    fn ensure_schema<'a>(
748        &'a self,
749        context: &'a UserContext,
750        _invocation: &'a teaql_runtime::SchemaInvocation,
751    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>> {
752        Box::pin(async move {
753            ensure_sqlite_schema_for(context).map_err(|err| RuntimeError::Schema(err.to_string()))
754        })
755    }
756}
757
758pub trait SqliteProviderExt {
759    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self;
760}
761
762impl SqliteProviderExt for UserContext {
763    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self {
764        self.insert_resource(SqliteDialect);
765        self.insert_resource(executor);
766        self.set_schema_provider(SqliteSchemaProvider);
767        self
768    }
769}
770
771#[derive(Clone)]
772pub struct SqliteIdSpaceGenerator {
773    executor: SqliteMutationExecutor,
774    table_name: String,
775}
776
777impl SqliteIdSpaceGenerator {
778    pub fn new(connection: Connection) -> Self {
779        Self::from_executor(SqliteMutationExecutor::from_connection(connection))
780    }
781
782    pub fn from_executor(executor: SqliteMutationExecutor) -> Self {
783        Self {
784            executor,
785            table_name: DEFAULT_ID_SPACE_TABLE.to_owned(),
786        }
787    }
788
789    pub fn with_table_name(mut self, table_name: impl Into<String>) -> Self {
790        self.table_name = table_name.into();
791        self
792    }
793
794    pub fn ensure_table(&self) -> Result<(), MutationExecutorError> {
795        self.executor.ensure_id_space_table(&self.table_name)
796    }
797
798    pub fn next_id(&self, entity: &str) -> Result<u64, MutationExecutorError> {
799        let entity = canonical_id_space_entity(entity);
800        let entity = entity.as_str();
801        self.ensure_table()?;
802        let table = quote_ident(&self.table_name);
803        let select_sql = format!("SELECT current_level FROM {table} WHERE type_name = ?");
804        let insert_sql = format!("INSERT INTO {table} (type_name, current_level) VALUES (?, 1)");
805        let update_sql = format!(
806            "UPDATE {table} SET current_level = ? WHERE type_name = ? AND current_level = ?"
807        );
808        for attempt in 1..=100 {
809            let connection = self.executor.lock()?;
810            let current = connection
811                .query_row(&select_sql, [entity], |row| row.get::<_, i64>(0))
812                .optional()?;
813            if let Some(current) = current {
814                let next = current.checked_add(1).ok_or_else(|| {
815                    MutationExecutorError::Bind(format!(
816                        "ID space overflow for {entity} on optimistic-lock attempt {attempt}"
817                    ))
818                })?;
819                if connection.execute(&update_sql, params![next, entity, current])? == 1 {
820                    return u64::try_from(next).map_err(|_| {
821                        MutationExecutorError::Bind(format!(
822                            "generated id {next} cannot be represented as u64"
823                        ))
824                    });
825                }
826            } else {
827                match connection.execute(&insert_sql, params![entity]) {
828                    Ok(1) => return Ok(1),
829                    Ok(changed) => {
830                        return Err(MutationExecutorError::Bind(format!(
831                            "ID space insert for {entity} changed {changed} rows"
832                        )));
833                    }
834                    Err(error)
835                        if error.sqlite_error_code()
836                            == Some(rusqlite::ErrorCode::ConstraintViolation) => {}
837                    Err(error) => return Err(error.into()),
838                }
839            }
840        }
841        Err(MutationExecutorError::Bind(format!(
842            "Unable to allocate ID for {entity} after 100 optimistic-lock attempts"
843        )))
844    }
845
846    pub fn ensure_floor(&self, entity: &str, floor: u64) -> Result<(), MutationExecutorError> {
847        let entity = canonical_id_space_entity(entity);
848        let entity = entity.as_str();
849        self.ensure_table()?;
850        let floor = i64::try_from(floor).map_err(|_| {
851            MutationExecutorError::Bind(format!("ID space floor {floor} for {entity} exceeds i64"))
852        })?;
853        let table = quote_ident(&self.table_name);
854        for _ in 1..=100 {
855            let connection = self.executor.lock()?;
856            let current = connection
857                .query_row(
858                    &format!("SELECT current_level FROM {table} WHERE type_name = ?"),
859                    [entity],
860                    |row| row.get::<_, i64>(0),
861                )
862                .optional()?;
863            match current {
864                Some(current) if current >= floor => return Ok(()),
865                Some(current) => {
866                    if connection.execute(
867                        &format!("UPDATE {table} SET current_level = ? WHERE type_name = ? AND current_level = ?"),
868                        params![floor, entity, current],
869                    )? == 1 { return Ok(()); }
870                }
871                None => match connection.execute(
872                    &format!("INSERT INTO {table}(type_name, current_level) VALUES (?, ?)"),
873                    params![entity, floor],
874                ) {
875                    Ok(1) => return Ok(()),
876                    Ok(_) => {}
877                    Err(error) if error.sqlite_error_code() == Some(rusqlite::ErrorCode::ConstraintViolation) => {}
878                    Err(error) => return Err(error.into()),
879                },
880            }
881        }
882        Err(MutationExecutorError::Bind(format!(
883            "Unable to synchronize ID space floor for {entity} after 100 optimistic-lock attempts"
884        )))
885    }
886}
887
888impl InternalIdGenerator for SqliteIdSpaceGenerator {
889    fn generate_id(&self, entity: &str) -> Result<u64, RuntimeError> {
890        self.next_id(entity)
891            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))
892    }
893
894    fn ensure_floor(&self, entity: &str, floor: u64) -> Result<(), RuntimeError> {
895        SqliteIdSpaceGenerator::ensure_floor(self, entity, floor)
896            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))
897    }
898}
899
900fn quote_ident(ident: &str) -> String {
901    quote_identifier_if_needed(ident, '"')
902}
903
904/// Strip wrapping identifier quotes from a SQL identifier.
905///
906/// SQLite `PRAGMA table_info` returns bare column names (e.g. `description`),
907/// but generated `PropertyDescriptor::column_name` may carry quotes
908/// (e.g. `"description"`) when the name is a reserved keyword.  This helper
909/// normalises the column name so the two can be compared correctly during
910/// schema migration.
911fn strip_identifier_quotes(ident: &str) -> &str {
912    let bytes = ident.as_bytes();
913    if bytes.len() >= 2 {
914        let (first, last) = (bytes[0], bytes[bytes.len() - 1]);
915        if (first == b'"' && last == b'"')
916            || (first == b'`' && last == b'`')
917            || (first == b'[' && last == b']')
918        {
919            return &ident[1..ident.len() - 1];
920        }
921    }
922    ident
923}
924
925fn bind_values(values: &[Value]) -> Result<Vec<SqliteValue>, MutationExecutorError> {
926    values.iter().map(bind_sqlite_value).collect()
927}
928
929fn bind_sqlite_value(value: &Value) -> Result<SqliteValue, MutationExecutorError> {
930    match value {
931        Value::Null => Ok(SqliteValue::Null),
932        Value::Bool(v) => Ok(SqliteValue::Integer(i64::from(*v))),
933        Value::I64(v) => Ok(SqliteValue::Integer(*v)),
934        Value::U64(v) => i64::try_from(*v)
935            .map(SqliteValue::Integer)
936            .map_err(|_| MutationExecutorError::Bind(format!("u64 value {v} exceeds i64 range"))),
937        Value::F64(v) => Ok(SqliteValue::Real(*v)),
938        // Bind the canonical numeric spelling. SQLite NUMERIC affinity keeps
939        // predicates and aggregates numeric; an application-only text prefix
940        // makes range comparisons silently return the wrong result.
941        Value::Decimal(v) => Ok(SqliteValue::Text(v.to_string())),
942        Value::Text(v) => Ok(SqliteValue::Text(v.clone())),
943        Value::Json(v) => Ok(SqliteValue::Text(v.to_string())),
944        Value::Date(v) => Ok(SqliteValue::Text(v.format("%Y-%m-%d").to_string())),
945        Value::Timestamp(v) => Ok(SqliteValue::Integer(v.0)),
946        Value::Object(_) => Err(MutationExecutorError::UnsupportedValue("object")),
947        Value::List(_) => Err(MutationExecutorError::UnsupportedValue("list")),
948        Value::TypedNull(_) => Ok(SqliteValue::Null),
949    }
950}
951
952#[derive(Debug, Clone)]
953struct ColumnInfo {
954    name: String,
955    decode_kind: SqliteDecodeKind,
956}
957
958#[derive(Debug, Clone, Copy, PartialEq, Eq)]
959enum SqliteDecodeKind {
960    Infer,
961    Bool,
962    Decimal,
963    Json,
964    Date,
965    Timestamp,
966    Text,
967}
968
969#[derive(Debug)]
970struct ColumnLayout {
971    columns: Arc<[ColumnInfo]>,
972    names: Arc<[String]>,
973}
974
975fn cached_column_layout(
976    cache: &Mutex<HashMap<String, Arc<ColumnLayout>>>,
977    sql: &str,
978    statement: &rusqlite::Statement<'_>,
979) -> Arc<ColumnLayout> {
980    if let Ok(cache) = cache.lock()
981        && let Some(layout) = cache.get(sql)
982    {
983        return layout.clone();
984    }
985
986    let columns: Arc<[ColumnInfo]> = statement_columns(statement).into();
987    let names = columns
988        .iter()
989        .map(|column| column.name.clone())
990        .collect::<Vec<_>>()
991        .into();
992    let layout = Arc::new(ColumnLayout { columns, names });
993    if let Ok(mut cache) = cache.lock() {
994        if cache.len() >= DEFAULT_COLUMN_LAYOUT_CACHE_CAPACITY {
995            cache.clear();
996        }
997        cache.insert(sql.to_owned(), layout.clone());
998    }
999    layout
1000}
1001
1002fn statement_columns(statement: &rusqlite::Statement<'_>) -> Vec<ColumnInfo> {
1003    statement
1004        .columns()
1005        .into_iter()
1006        .map(|column| ColumnInfo {
1007            name: column.name().to_owned(),
1008            decode_kind: sqlite_decode_kind(column.decl_type()),
1009        })
1010        .collect()
1011}
1012
1013fn sqlite_decode_kind(decl_type: Option<&str>) -> SqliteDecodeKind {
1014    let Some(decl_type) = decl_type else {
1015        return SqliteDecodeKind::Infer;
1016    };
1017    let base = decl_type.split('(').next().unwrap_or(decl_type).trim();
1018    if base.eq_ignore_ascii_case("BOOLEAN") || base.eq_ignore_ascii_case("BOOL") {
1019        SqliteDecodeKind::Bool
1020    } else if base.eq_ignore_ascii_case("NUMERIC") || base.eq_ignore_ascii_case("DECIMAL") {
1021        SqliteDecodeKind::Decimal
1022    } else if base.eq_ignore_ascii_case("JSON") {
1023        SqliteDecodeKind::Json
1024    } else if base.eq_ignore_ascii_case("DATE") {
1025        SqliteDecodeKind::Date
1026    } else if base.eq_ignore_ascii_case("TIMESTAMP") || base.eq_ignore_ascii_case("DATETIME") {
1027        SqliteDecodeKind::Timestamp
1028    } else if ["TEXT", "VARCHAR", "CHAR", "CLOB"]
1029        .iter()
1030        .any(|v| base.eq_ignore_ascii_case(v))
1031    {
1032        SqliteDecodeKind::Text
1033    } else {
1034        SqliteDecodeKind::Infer
1035    }
1036}
1037
1038fn decode_sqlite_values(
1039    row: &Row<'_>,
1040    columns: &[ColumnInfo],
1041) -> Result<Vec<Value>, MutationExecutorError> {
1042    let mut values = Vec::with_capacity(columns.len());
1043    for (index, column) in columns.iter().enumerate() {
1044        let value_ref = row.get_ref(index)?;
1045        let value = match value_ref {
1046            ValueRef::Null => Value::Null,
1047            ValueRef::Integer(value) => decode_sqlite_integer(value, column),
1048            ValueRef::Real(value) => Value::F64(value),
1049            ValueRef::Text(value) => decode_sqlite_text(value, column)?,
1050            ValueRef::Blob(_) => {
1051                return Err(MutationExecutorError::UnsupportedColumnType(
1052                    "BLOB".to_owned(),
1053                ));
1054            }
1055        };
1056        values.push(value);
1057    }
1058    Ok(values)
1059}
1060
1061fn decode_sqlite_integer(value: i64, column: &ColumnInfo) -> Value {
1062    match column.decode_kind {
1063        SqliteDecodeKind::Bool => Value::Bool(value != 0),
1064        _ => Value::I64(value),
1065    }
1066}
1067
1068fn decode_sqlite_text(value: &[u8], column: &ColumnInfo) -> Result<Value, MutationExecutorError> {
1069    let value = std::str::from_utf8(value)
1070        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite text: {err}")))?;
1071    match column.decode_kind {
1072        SqliteDecodeKind::Decimal => Decimal::from_str(value)
1073            .map(Value::Decimal)
1074            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite decimal: {err}"))),
1075        SqliteDecodeKind::Json => serde_json::from_str(value).map(Value::Json).map_err(|err| {
1076            MutationExecutorError::Bind(format!("invalid sqlite json value: {err}"))
1077        }),
1078        SqliteDecodeKind::Date => NaiveDate::parse_from_str(value, "%Y-%m-%d")
1079            .map(Value::Date)
1080            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite date: {err}"))),
1081        SqliteDecodeKind::Timestamp => parse_sqlite_timestamp(value),
1082        SqliteDecodeKind::Text | SqliteDecodeKind::Bool => Ok(Value::Text(value.to_owned())),
1083        SqliteDecodeKind::Infer => infer_sqlite_text(value),
1084    }
1085}
1086
1087fn infer_sqlite_text(value: &str) -> Result<Value, MutationExecutorError> {
1088    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
1089        return Ok(Value::Date(date));
1090    }
1091    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
1092        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
1093            timestamp.timestamp_millis(),
1094        )));
1095    }
1096    if let Ok(timestamp) = NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S") {
1097        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
1098            timestamp.and_utc().timestamp_millis(),
1099        )));
1100    }
1101    Ok(Value::Text(value.to_owned()))
1102}
1103
1104fn parse_sqlite_timestamp(value: &str) -> Result<Value, MutationExecutorError> {
1105    if let Some(timestamp) = parse_fixed_sqlite_timestamp(value) {
1106        return Ok(Value::Timestamp(teaql_core::time::Timestamp(timestamp)));
1107    }
1108    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
1109        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
1110            timestamp.timestamp_millis(),
1111        )));
1112    }
1113    if let Ok(timestamp) = DateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S%.f%#z") {
1114        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
1115            timestamp.timestamp_millis(),
1116        )));
1117    }
1118    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
1119        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
1120            date.and_hms_opt(0, 0, 0)
1121                .unwrap_or_default()
1122                .and_utc()
1123                .timestamp_millis(),
1124        )));
1125    }
1126    NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S%.f")
1127        .map(|timestamp| {
1128            Value::Timestamp(teaql_core::time::Timestamp(
1129                timestamp.and_utc().timestamp_millis(),
1130            ))
1131        })
1132        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite timestamp: {err}")))
1133}
1134
1135fn parse_fixed_sqlite_timestamp(value: &str) -> Option<i64> {
1136    let bytes = value.as_bytes();
1137    if bytes.len() < 19
1138        || bytes.get(4) != Some(&b'-')
1139        || bytes.get(7) != Some(&b'-')
1140        || !matches!(bytes.get(10), Some(b' ') | Some(b'T'))
1141        || bytes.get(13) != Some(&b':')
1142        || bytes.get(16) != Some(&b':')
1143    {
1144        return None;
1145    }
1146    let digits = |start: usize, len: usize| -> Option<u32> {
1147        bytes
1148            .get(start..start + len)?
1149            .iter()
1150            .try_fold(0_u32, |value, byte| {
1151                byte.is_ascii_digit()
1152                    .then_some(value * 10 + u32::from(*byte - b'0'))
1153            })
1154    };
1155    let date = NaiveDate::from_ymd_opt(
1156        i32::try_from(digits(0, 4)?).ok()?,
1157        digits(5, 2)?,
1158        digits(8, 2)?,
1159    )?;
1160    let hour = digits(11, 2)?;
1161    let minute = digits(14, 2)?;
1162    let second = digits(17, 2)?;
1163    let mut cursor = 19;
1164    let mut nanos = 0_u32;
1165    if bytes.get(cursor) == Some(&b'.') {
1166        cursor += 1;
1167        let fraction_start = cursor;
1168        while bytes.get(cursor).is_some_and(u8::is_ascii_digit) {
1169            if cursor - fraction_start < 9 {
1170                nanos = nanos * 10 + u32::from(bytes[cursor] - b'0');
1171            }
1172            cursor += 1;
1173        }
1174        let kept = (cursor - fraction_start).min(9);
1175        if kept == 0 {
1176            return None;
1177        }
1178        nanos *= 10_u32.pow(u32::try_from(9 - kept).ok()?);
1179    }
1180    let datetime = date.and_hms_nano_opt(hour, minute, second, nanos)?;
1181    let offset_seconds = match bytes.get(cursor..) {
1182        Some([]) | Some([b'Z']) | Some([b'z']) => 0,
1183        Some([sign @ (b'+' | b'-'), hour_1, hour_2]) => {
1184            signed_offset(*sign, [*hour_1, *hour_2], *b"00")?
1185        }
1186        Some([sign @ (b'+' | b'-'), hour_1, hour_2, minute_1, minute_2]) => {
1187            signed_offset(*sign, [*hour_1, *hour_2], [*minute_1, *minute_2])?
1188        }
1189        Some(
1190            [
1191                sign @ (b'+' | b'-'),
1192                hour_1,
1193                hour_2,
1194                b':',
1195                minute_1,
1196                minute_2,
1197            ],
1198        ) => signed_offset(*sign, [*hour_1, *hour_2], [*minute_1, *minute_2])?,
1199        _ => return None,
1200    };
1201    FixedOffset::east_opt(offset_seconds)?
1202        .from_local_datetime(&datetime)
1203        .single()
1204        .map(|timestamp| timestamp.timestamp_millis())
1205}
1206
1207fn signed_offset(sign: u8, hours: [u8; 2], minutes: [u8; 2]) -> Option<i32> {
1208    let pair = |digits: [u8; 2]| {
1209        digits
1210            .iter()
1211            .all(u8::is_ascii_digit)
1212            .then_some(i32::from(digits[0] - b'0') * 10 + i32::from(digits[1] - b'0'))
1213    };
1214    let hours = pair(hours)?;
1215    let minutes = pair(minutes)?;
1216    if hours > 23 || minutes > 59 {
1217        return None;
1218    }
1219    let seconds = hours * 3600 + minutes * 60;
1220    Some(if sign == b'-' { -seconds } else { seconds })
1221}
1222
1223#[cfg(test)]
1224mod tests {
1225    use super::*;
1226    use futures_util::StreamExt;
1227    use teaql_core::{DeleteCommand, Record, RecoverCommand};
1228    use teaql_macros::TeaqlEntity;
1229    use teaql_runtime::InMemoryMetadataStore;
1230
1231    #[test]
1232    fn ensure_schema_registers_soundex_idempotently() {
1233        let executor =
1234            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1235        executor.ensure_schema(&SqliteDialect, &[]).unwrap();
1236        executor.ensure_schema(&SqliteDialect, &[]).unwrap();
1237        let connection = executor.connection();
1238        let guard = connection.lock().unwrap();
1239        let encoded: String = guard
1240            .query_row("SELECT soundex('Robert')", [], |row| row.get(0))
1241            .unwrap();
1242        let matches: i64 = guard
1243            .query_row("SELECT soundex('Robert') = soundex('Rupert')", [], |row| {
1244                row.get(0)
1245            })
1246            .unwrap();
1247        let empty: String = guard
1248            .query_row("SELECT soundex(NULL)", [], |row| row.get(0))
1249            .unwrap();
1250        assert_eq!(encoded, "R163");
1251        assert_eq!(matches, 1);
1252        assert_eq!(empty, "?000");
1253    }
1254
1255    #[test]
1256    fn streaming_sql_yields_bounded_chunks_and_releases_cursor_on_drop() {
1257        let connection = Connection::open_in_memory().unwrap();
1258        connection
1259            .execute_batch(
1260                "CREATE TABLE stream_fixture(id INTEGER);\
1261                 INSERT INTO stream_fixture VALUES (1), (2), (3), (4), (5);",
1262            )
1263            .unwrap();
1264        let executor = SqliteMutationExecutor::from_connection(connection);
1265        let query = CompiledQuery {
1266            sql: "SELECT id FROM stream_fixture ORDER BY id".to_owned(),
1267            params: vec![],
1268            comment: None,
1269        };
1270        let mut stream = teaql_sql::StreamingSqlTransport::stream_sql(&executor, query.clone(), 2);
1271        let sizes = futures_executor::block_on(async {
1272            let mut result = Vec::new();
1273            while let Some(chunk) = stream.next().await {
1274                result.push(chunk.unwrap().rows.len());
1275            }
1276            result
1277        });
1278        assert_eq!(sizes, vec![2, 2, 1]);
1279
1280        let mut early = teaql_sql::StreamingSqlTransport::stream_sql(&executor, query, 2);
1281        assert_eq!(
1282            futures_executor::block_on(early.next())
1283                .unwrap()
1284                .unwrap()
1285                .rows
1286                .len(),
1287            2
1288        );
1289        drop(early);
1290        let count: i64 = executor
1291            .connection()
1292            .lock()
1293            .unwrap()
1294            .query_row("SELECT count(*) FROM stream_fixture", [], |row| row.get(0))
1295            .unwrap();
1296        assert_eq!(count, 5);
1297    }
1298
1299    #[test]
1300    fn decimal_bind_is_numeric_and_comparable() {
1301        let value =
1302            bind_sqlite_value(&Value::Decimal(Decimal::from_str("123.450").unwrap())).unwrap();
1303        assert_eq!(value, SqliteValue::Text("123.450".to_owned()));
1304        let connection = Connection::open_in_memory().unwrap();
1305        let matches: i64 = connection
1306            .query_row(
1307                "SELECT 1 WHERE CAST(? AS NUMERIC) BETWEEN 120 AND 130",
1308                [value],
1309                |row| row.get(0),
1310            )
1311            .unwrap();
1312        assert_eq!(matches, 1);
1313    }
1314
1315    #[test]
1316    fn temporal_debug_sql_is_executable_and_matches_prepared_storage() {
1317        let connection = Connection::open_in_memory().unwrap();
1318        connection
1319            .execute_batch(
1320                "CREATE TABLE temporal_fixture (id INTEGER PRIMARY KEY, d DATE, t TIMESTAMP)",
1321            )
1322            .unwrap();
1323        let query = CompiledQuery {
1324            sql: "INSERT INTO temporal_fixture VALUES (?, ?, ?)".to_owned(),
1325            params: vec![
1326                Value::I64(1),
1327                Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 2, 29).unwrap()),
1328                Value::Timestamp(teaql_core::time::Timestamp(1_787_110_200_123)),
1329            ],
1330            comment: None,
1331        };
1332        let values = bind_values(&query.params).unwrap();
1333        connection
1334            .execute(&query.sql, rusqlite::params_from_iter(values))
1335            .unwrap();
1336        connection
1337            .execute(
1338                &query
1339                    .debug_sql(teaql_sql::DatabaseKind::Sqlite)
1340                    .replace("VALUES (1,", "VALUES (2,"),
1341                [],
1342            )
1343            .unwrap();
1344
1345        let equal_count: i64 = connection.query_row(
1346            "SELECT count(*) FROM temporal_fixture a JOIN temporal_fixture b ON a.d=b.d AND a.t=b.t WHERE a.id=1 AND b.id=2",
1347            [], |row| row.get(0),
1348        ).unwrap();
1349        let storage_type: String = connection
1350            .query_row(
1351                "SELECT typeof(t) FROM temporal_fixture WHERE id=1",
1352                [],
1353                |row| row.get(0),
1354            )
1355            .unwrap();
1356        assert_eq!(equal_count, 1);
1357        assert_eq!(storage_type, "integer");
1358    }
1359
1360    fn entity() -> EntityDescriptor {
1361        EntityDescriptor::new("Order")
1362            .table_name("orders")
1363            .property(
1364                PropertyDescriptor::new("id", DataType::U64)
1365                    .column_name("id")
1366                    .id()
1367                    .not_null(),
1368            )
1369            .property(
1370                PropertyDescriptor::new("version", DataType::I64)
1371                    .column_name("version")
1372                    .version()
1373                    .not_null(),
1374            )
1375            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
1376    }
1377
1378    fn order_line_entity() -> EntityDescriptor {
1379        EntityDescriptor::new("OrderLine")
1380            .table_name("order_line")
1381            .property(
1382                PropertyDescriptor::new("id", DataType::U64)
1383                    .column_name("id")
1384                    .id()
1385                    .not_null(),
1386            )
1387            .property(
1388                PropertyDescriptor::new("order_id", DataType::U64)
1389                    .column_name("order_id")
1390                    .not_null(),
1391            )
1392            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
1393    }
1394
1395    fn complete_query_record_entity() -> EntityDescriptor {
1396        EntityDescriptor::new("QueryRecord")
1397            .table_name("query_record_scalar")
1398            .property(PropertyDescriptor::new("id", DataType::U64).id().not_null())
1399            .property(PropertyDescriptor::new("required_text", DataType::Text))
1400            .property(PropertyDescriptor::new("optional_text", DataType::Text))
1401            .property(PropertyDescriptor::new("required_integer", DataType::I64))
1402            .property(PropertyDescriptor::new("optional_long", DataType::I64))
1403            .property(PropertyDescriptor::new(
1404                "required_decimal",
1405                DataType::Decimal,
1406            ))
1407            .property(PropertyDescriptor::new("required_float", DataType::F64))
1408            .property(PropertyDescriptor::new("required_double", DataType::F64))
1409            .property(PropertyDescriptor::new("required_date", DataType::Date))
1410            .property(PropertyDescriptor::new("required_time", DataType::I64))
1411            .property(PropertyDescriptor::new(
1412                "required_timestamp",
1413                DataType::Timestamp,
1414            ))
1415            .property(PropertyDescriptor::new("active", DataType::Bool))
1416            .property(PropertyDescriptor::new("reviewed", DataType::Bool))
1417            .property(
1418                PropertyDescriptor::new("version", DataType::I64)
1419                    .version()
1420                    .not_null(),
1421            )
1422    }
1423
1424    #[test]
1425    fn complete_scalar_fixture_including_nullable_boolean_executes_on_sqlite() {
1426        let executor = SqliteMutationExecutor::from_connection(
1427            Connection::open_in_memory().expect("open SQLite fixture"),
1428        );
1429        executor
1430            .connection()
1431            .lock()
1432            .expect("lock SQLite fixture")
1433            .execute_batch("CREATE TABLE query_record_scalar (\
1434                    id INTEGER PRIMARY KEY, required_text TEXT, optional_text TEXT,\
1435                    required_integer INTEGER, optional_long INTEGER, required_decimal NUMERIC,\
1436                    required_float REAL, required_double REAL, required_date DATE,\
1437                    required_time INTEGER, required_timestamp TIMESTAMP,\
1438                    active BOOLEAN, reviewed BOOLEAN, version INTEGER);\
1439                    INSERT INTO query_record_scalar VALUES \
1440                    (1,'Alpha','optional',42,42000000000,42.125,42.5,42.75,'2026-08-29',34200000,1777632600000,1,0,1),\
1441                    (2,'Beta',NULL,7,NULL,7.500,7.5,7.75,'2026-08-30',36000000,1777720400000,0,NULL,1),\
1442                    (3,'Gamma','tail',99,99000000000,99.875,99.5,99.75,'2026-08-31',37800000,1777808200000,1,1,1)")
1443            .expect("seed complete scalar fixture");
1444        let entity = complete_query_record_entity();
1445        let ids = |expr: Expr| {
1446            let query = SelectQuery::new("QueryRecord")
1447                .project("id")
1448                .filter(expr)
1449                .order_asc("id");
1450            executor
1451                .fetch_all_compact(&SqliteDialect.compile_select(&entity, &query).unwrap())
1452                .expect("execute scalar predicate")
1453                .into_iter()
1454                .map(|row| row.get("id").cloned().expect("projected id"))
1455                .collect::<Vec<_>>()
1456        };
1457        assert_eq!(ids(Expr::eq("required_text", "Alpha")), vec![Value::I64(1)]);
1458        assert_eq!(
1459            ids(Expr::ne("required_text", "Alpha")),
1460            vec![Value::I64(2), Value::I64(3)]
1461        );
1462        assert_eq!(
1463            ids(Expr::in_list(
1464                "required_text",
1465                [Value::from("Alpha"), Value::from("Gamma")]
1466            )),
1467            vec![Value::I64(1), Value::I64(3)]
1468        );
1469        assert_eq!(
1470            ids(Expr::contain("required_text", "et")),
1471            vec![Value::I64(2)]
1472        );
1473        assert_eq!(
1474            ids(Expr::between("required_integer", 40_i64, 100_i64)),
1475            vec![Value::I64(1), Value::I64(3)]
1476        );
1477        assert_eq!(
1478            ids(Expr::gt("required_decimal", Decimal::from(50))),
1479            vec![Value::I64(3)]
1480        );
1481        assert_eq!(
1482            ids(Expr::lte("required_float", 7.5_f64)),
1483            vec![Value::I64(2)]
1484        );
1485        assert_eq!(
1486            ids(Expr::gte("required_double", 99.75_f64)),
1487            vec![Value::I64(3)]
1488        );
1489        assert_eq!(
1490            ids(Expr::between(
1491                "required_date",
1492                NaiveDate::from_ymd_opt(2026, 8, 30).unwrap(),
1493                NaiveDate::from_ymd_opt(2026, 8, 31).unwrap(),
1494            )),
1495            vec![Value::I64(2), Value::I64(3)]
1496        );
1497        assert_eq!(
1498            ids(Expr::gt("required_time", 36_000_000_i64)),
1499            vec![Value::I64(3)]
1500        );
1501        assert_eq!(
1502            ids(Expr::lt(
1503                "required_timestamp",
1504                teaql_core::time::Timestamp(1_777_750_000_000)
1505            )),
1506            vec![Value::I64(1), Value::I64(2)]
1507        );
1508        assert_eq!(ids(Expr::is_null("optional_text")), vec![Value::I64(2)]);
1509        assert_eq!(
1510            ids(Expr::is_not_null("optional_long")),
1511            vec![Value::I64(1), Value::I64(3)]
1512        );
1513        assert_eq!(ids(Expr::eq("active", false)), vec![Value::I64(2)]);
1514        assert_eq!(ids(Expr::eq("reviewed", true)), vec![Value::I64(3)]);
1515        assert_eq!(ids(Expr::eq("reviewed", false)), vec![Value::I64(1)]);
1516        assert_eq!(ids(Expr::is_null("reviewed")), vec![Value::I64(2)]);
1517    }
1518
1519    #[test]
1520    fn relation_subqueries_execute_positive_and_negative_predicates_on_sqlite() {
1521        let executor = SqliteMutationExecutor::from_connection(
1522            Connection::open_in_memory().expect("open SQLite fixture"),
1523        );
1524        executor
1525            .connection()
1526            .lock()
1527            .expect("lock SQLite fixture")
1528            .execute_batch(
1529                "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name TEXT);\
1530                 CREATE TABLE order_line (id INTEGER PRIMARY KEY, order_id INTEGER, name TEXT);\
1531                 INSERT INTO orders VALUES (1, 1, 'first'), (2, 1, 'second'), (3, 1, 'third');\
1532                 INSERT INTO order_line VALUES\
1533                    (10, 1, 'priority'), (11, 1, 'ordinary'), (12, 2, 'ordinary'),\
1534                    (13, NULL, 'orphan');",
1535            )
1536            .expect("seed relation fixture");
1537
1538        let matching_lines = SelectQuery::new("OrderLine").filter(Expr::eq("name", "priority"));
1539        let positive = SelectQuery::new("Order")
1540            .project("id")
1541            .filter(Expr::in_subquery(
1542                "id",
1543                order_line_entity(),
1544                matching_lines.clone(),
1545                "order_id",
1546            ))
1547            .order_asc("id");
1548        let negative = SelectQuery::new("Order")
1549            .project("id")
1550            .filter(Expr::not_in_subquery(
1551                "id",
1552                order_line_entity(),
1553                matching_lines,
1554                "order_id",
1555            ))
1556            .order_asc("id");
1557
1558        let ids = |rows: Vec<CompactRow>| {
1559            rows.into_iter()
1560                .map(|row| row.get("id").cloned().expect("projected id"))
1561                .collect::<Vec<_>>()
1562        };
1563        let order_ids = |query: SelectQuery| {
1564            ids(executor
1565                .fetch_all_compact(&SqliteDialect.compile_select(&entity(), &query).unwrap())
1566                .expect("execute order relation predicate"))
1567        };
1568        let line_ids = |query: SelectQuery| {
1569            ids(executor
1570                .fetch_all_compact(
1571                    &SqliteDialect
1572                        .compile_select(&order_line_entity(), &query)
1573                        .unwrap(),
1574                )
1575                .expect("execute line relation predicate"))
1576        };
1577
1578        // Reverse relation: typed child matching and its negative form.
1579        assert_eq!(order_ids(positive), vec![Value::I64(1)]);
1580        assert_eq!(order_ids(negative), vec![Value::I64(2), Value::I64(3)]);
1581
1582        // Forward relation identity state keeps NULL distinct from a known FK.
1583        assert_eq!(
1584            line_ids(
1585                SelectQuery::new("OrderLine")
1586                    .project("id")
1587                    .filter(Expr::is_not_null("order_id"))
1588                    .order_asc("id")
1589            ),
1590            vec![Value::I64(10), Value::I64(11), Value::I64(12)]
1591        );
1592        assert_eq!(
1593            line_ids(
1594                SelectQuery::new("OrderLine")
1595                    .project("id")
1596                    .filter(Expr::is_null("order_id"))
1597                    .order_asc("id")
1598            ),
1599            vec![Value::I64(13)]
1600        );
1601
1602        // Forward nested matching. SQL NOT IN deliberately excludes the NULL
1603        // foreign key; callers use IsUnknown when they want orphan rows.
1604        let first_order = SelectQuery::new("Order").filter(Expr::eq("name", "first"));
1605        assert_eq!(
1606            line_ids(
1607                SelectQuery::new("OrderLine")
1608                    .project("id")
1609                    .filter(Expr::in_subquery(
1610                        "order_id",
1611                        entity(),
1612                        first_order.clone(),
1613                        "id",
1614                    ))
1615                    .order_asc("id")
1616            ),
1617            vec![Value::I64(10), Value::I64(11)]
1618        );
1619        assert_eq!(
1620            line_ids(
1621                SelectQuery::new("OrderLine")
1622                    .project("id")
1623                    .filter(Expr::not_in_subquery(
1624                        "order_id",
1625                        entity(),
1626                        first_order,
1627                        "id",
1628                    ))
1629                    .order_asc("id")
1630            ),
1631            vec![Value::I64(12)]
1632        );
1633
1634        // Reverse existence/non-existence without an additional child filter.
1635        let all_lines = SelectQuery::new("OrderLine");
1636        assert_eq!(
1637            order_ids(
1638                SelectQuery::new("Order")
1639                    .project("id")
1640                    .filter(Expr::in_subquery(
1641                        "id",
1642                        order_line_entity(),
1643                        all_lines.clone(),
1644                        "order_id",
1645                    ))
1646                    .order_asc("id")
1647            ),
1648            vec![Value::I64(1), Value::I64(2)]
1649        );
1650        assert_eq!(
1651            order_ids(
1652                SelectQuery::new("Order")
1653                    .project("id")
1654                    .filter(Expr::not_in_subquery(
1655                        "id",
1656                        order_line_entity(),
1657                        all_lines,
1658                        "order_id",
1659                    ))
1660                    .order_asc("id")
1661            ),
1662            vec![Value::I64(3)]
1663        );
1664    }
1665
1666    #[allow(dead_code)]
1667    #[derive(Debug, PartialEq, TeaqlEntity)]
1668    #[teaql(entity = "FeatureFlag", table = "feature_flags")]
1669    struct FeatureFlagRow {
1670        #[teaql(id)]
1671        id: u64,
1672        #[teaql(version)]
1673        version: i64,
1674        enabled: bool,
1675        optional_enabled: Option<bool>,
1676    }
1677
1678    fn feature_flag_record(enabled: Value, optional_enabled: Value) -> Record {
1679        Record::from([
1680            ("id".to_owned(), Value::U64(1)),
1681            ("version".to_owned(), Value::I64(1)),
1682            ("enabled".to_owned(), enabled),
1683            ("optional_enabled".to_owned(), optional_enabled),
1684        ])
1685    }
1686
1687    #[test]
1688    fn sqlite_dialect_compiles_mutations_and_schema() {
1689        assert!(SqliteDialect.prefers_small_parent_relation_probes());
1690        let insert = SqliteDialect
1691            .compile_insert(
1692                &entity(),
1693                &InsertCommand::new("Order")
1694                    .value("id", 1_u64)
1695                    .value("name", "A"),
1696            )
1697            .unwrap();
1698        assert_eq!(insert.sql, "INSERT INTO orders (id, name) VALUES (?, ?)");
1699
1700        let update = SqliteDialect
1701            .compile_update(
1702                &entity(),
1703                &UpdateCommand::new("Order", 1_u64)
1704                    .expected_version(3)
1705                    .value("name", "B"),
1706            )
1707            .unwrap();
1708        assert_eq!(
1709            update.sql,
1710            "UPDATE orders SET name = ?, version = ? WHERE id = ? AND version = ?"
1711        );
1712
1713        let delete = SqliteDialect
1714            .compile_delete(
1715                &entity(),
1716                &DeleteCommand::new("Order", 1_u64).expected_version(3),
1717            )
1718            .unwrap();
1719        let recover = SqliteDialect
1720            .compile_recover(&entity(), &RecoverCommand::new("Order", 1_u64, -4))
1721            .unwrap();
1722        assert_eq!(
1723            delete.sql,
1724            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
1725        );
1726        assert_eq!(
1727            recover.sql,
1728            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
1729        );
1730
1731        let create = SqliteDialect.compile_create_table(&entity()).unwrap();
1732        assert_eq!(
1733            create,
1734            "CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY NOT NULL, version INTEGER NOT NULL, name VARCHAR(255))"
1735        );
1736    }
1737
1738    #[test]
1739    fn column_layout_cache_uses_parameterized_sql_not_comments() {
1740        let connection = Connection::open_in_memory().unwrap();
1741        connection
1742            .execute("CREATE TABLE sample (id INTEGER, enabled BOOLEAN)", [])
1743            .unwrap();
1744        connection
1745            .execute("INSERT INTO sample (id, enabled) VALUES (1, 1)", [])
1746            .unwrap();
1747        let executor = SqliteMutationExecutor::from_connection(connection);
1748        let mut first = CompiledQuery {
1749            sql: "SELECT id, enabled FROM sample WHERE id = ?".to_owned(),
1750            params: vec![Value::I64(1)],
1751            comment: Some("first purpose".to_owned()),
1752        };
1753        let rows = executor.fetch_all_compact(&first).unwrap();
1754        assert_eq!(rows[0].get("enabled"), Some(&Value::Bool(true)));
1755
1756        first.comment = Some("different purpose".to_owned());
1757        executor.fetch_all_compact(&first).unwrap();
1758
1759        assert_eq!(executor.column_layout_cache.lock().unwrap().len(), 1);
1760    }
1761
1762    #[test]
1763    fn sqlite_executor_ensures_schema_and_roundtrips_rows() {
1764        let executor =
1765            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1766        let entity = entity();
1767        let mut context = UserContext::new()
1768            .with_metadata(InMemoryMetadataStore::new().with_entity(entity.clone()));
1769
1770        context.use_sqlite_provider(executor.clone());
1771        ensure_sqlite_schema_for(&context).unwrap();
1772
1773        let insert = SqliteDialect
1774            .compile_insert(
1775                &entity,
1776                &InsertCommand::new("Order")
1777                    .value("id", 1_u64)
1778                    .value("version", 1_i64)
1779                    .value("name", "draft"),
1780            )
1781            .unwrap();
1782        assert_eq!(executor.execute(&insert).unwrap(), 1);
1783
1784        let select = SqliteDialect
1785            .compile_select(
1786                &entity,
1787                &SelectQuery::new("Order")
1788                    .filter(Expr::eq("id", 1_u64))
1789                    .order_asc("id"),
1790            )
1791            .unwrap();
1792        let rows = executor.fetch_all_compact(&select).unwrap();
1793        assert_eq!(rows.len(), 1);
1794        assert_eq!(rows[0].get("id"), Some(&Value::I64(1)));
1795        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
1796        assert_eq!(rows[0].get("name"), Some(&Value::Text("draft".to_owned())));
1797    }
1798
1799    #[test]
1800    fn physical_schema_never_interprets_generated_bootstrap_graphs() {
1801        let executor =
1802            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1803        let entity = entity();
1804        let mut context = UserContext::new()
1805            .with_metadata(InMemoryMetadataStore::new().with_entity(entity.clone()));
1806        context.set_root_graphs(vec![
1807            GraphNode::new("Order")
1808                .value("id", 1_u64)
1809                .value("version", 1_i64)
1810                .value("name", "module seed"),
1811        ]);
1812        context.use_sqlite_provider(executor.clone());
1813
1814        ensure_sqlite_physical_schema_for(&context).unwrap();
1815
1816        let select = SqliteDialect
1817            .compile_select(
1818                &entity,
1819                &SelectQuery::new("Order").filter(Expr::eq("id", 1_u64)),
1820            )
1821            .unwrap();
1822        let rows = executor.fetch_all_compact(&select).unwrap();
1823        assert!(rows.is_empty());
1824    }
1825
1826    #[test]
1827    fn public_schema_boundary_rejects_legacy_provider_owned_bootstrap_graphs() {
1828        let executor =
1829            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1830        let entity = entity();
1831        let mut context = UserContext::new()
1832            .with_metadata(InMemoryMetadataStore::new().with_entity(entity.clone()));
1833        context.set_initial_graphs(vec![
1834            GraphNode::new("Order")
1835                .value("id", 1001_u64)
1836                .value("version", 1_i64)
1837                .value("name", "red"),
1838        ]);
1839        context.use_sqlite_provider(executor.clone());
1840        let error = ensure_sqlite_schema_for(&context).unwrap_err();
1841        assert!(error.to_string().contains(
1842            "generated root/constant bootstrap must use the typed RuntimeModule callback"
1843        ));
1844        let count: i64 = executor
1845            .lock()
1846            .unwrap()
1847            .query_row("SELECT COUNT(*) FROM orders", [], |row| row.get(0))
1848            .unwrap();
1849        assert_eq!(count, 0);
1850    }
1851
1852    #[test]
1853    fn sqlite_executes_partitioned_relation_limit_per_parent() {
1854        let executor =
1855            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1856        let entity = order_line_entity();
1857        executor.ensure_schema(&SqliteDialect, &[&entity]).unwrap();
1858
1859        for order_id in [11_u64, 12_u64] {
1860            for index in 1_u64..=5 {
1861                let id = order_id * 100 + index;
1862                let insert = SqliteDialect
1863                    .compile_insert(
1864                        &entity,
1865                        &InsertCommand::new("OrderLine")
1866                            .value("id", id)
1867                            .value("order_id", order_id)
1868                            .value("name", format!("line-{id}")),
1869                    )
1870                    .unwrap();
1871                executor.execute(&insert).unwrap();
1872            }
1873        }
1874
1875        let query = SelectQuery::new("OrderLine")
1876            .project("id")
1877            .project("order_id")
1878            .order_desc("id")
1879            .limit(3)
1880            .partition_by("order_id");
1881        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1882        let rows = executor.fetch_all_compact(&compiled).unwrap();
1883
1884        assert_eq!(rows.len(), 6);
1885        for order_id in [11_i64, 12_i64] {
1886            let ids = rows
1887                .iter()
1888                .filter(|row| row.get("order_id") == Some(&Value::I64(order_id)))
1889                .filter_map(|row| row.get("id").cloned())
1890                .collect::<Vec<_>>();
1891            assert_eq!(
1892                ids,
1893                vec![
1894                    Value::I64(order_id * 100 + 5),
1895                    Value::I64(order_id * 100 + 4),
1896                    Value::I64(order_id * 100 + 3),
1897                ]
1898            );
1899        }
1900    }
1901
1902    #[test]
1903    fn topn_005_007_window_and_probes_preserve_results_and_predicates() {
1904        futures_executor::block_on(async {
1905            #[derive(Clone)]
1906            struct FixedSchema(Arc<EntityDescriptor>);
1907
1908            impl teaql_data_service::SchemaProvider for FixedSchema {
1909                fn get_entity(&self, name: &str) -> Option<Arc<EntityDescriptor>> {
1910                    (name == self.0.name).then(|| self.0.clone())
1911                }
1912            }
1913
1914            let transport =
1915                SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1916            let entity = Arc::new(order_line_entity());
1917            transport
1918                .ensure_schema(&SqliteDialect, &[entity.as_ref()])
1919                .unwrap();
1920
1921            for order_id in [11_u64, 12_u64, 13_u64] {
1922                for index in 1_u64..=5 {
1923                    let id = order_id * 100 + index;
1924                    let name = if index == 4 { "excluded" } else { "visible" };
1925                    let insert = SqliteDialect
1926                        .compile_insert(
1927                            &entity,
1928                            &InsertCommand::new("OrderLine")
1929                                .value("id", id)
1930                                .value("order_id", order_id)
1931                                .value("name", name),
1932                        )
1933                        .unwrap();
1934                    transport.execute(&insert).unwrap();
1935                }
1936            }
1937
1938            let executor = teaql_sql::SqlDataServiceExecutor::new(
1939                SqliteDialect,
1940                transport,
1941                FixedSchema(entity),
1942            );
1943            let base = SelectQuery::new("OrderLine")
1944                .project("id")
1945                .project("order_id")
1946                .project("name")
1947                .filter(Expr::in_list("order_id", [Value::U64(11), Value::U64(12)]))
1948                .and_filter(Expr::eq("name", "visible"))
1949                .order_desc("id")
1950                .limit(3)
1951                .partition_by("order_id");
1952            let execute = |query| {
1953                teaql_data_service::QueryExecutor::query(
1954                    &executor,
1955                    teaql_data_service::QueryRequest {
1956                        query,
1957                        trace_chain: Vec::new(),
1958                        comment: Some("TOPN plan equivalence".to_owned()),
1959                        capture_debug_query: false,
1960                        capture_execution_metadata: false,
1961                    },
1962                )
1963            };
1964
1965            let probes = execute(base.clone()).await.unwrap().rows;
1966            let window = execute(base.top_n_probe_parent_threshold(0))
1967                .await
1968                .unwrap()
1969                .rows;
1970            let children_of = |rows: &[CompactRow], parent: i64| {
1971                rows.iter()
1972                    .filter(|row| row.get("order_id") == Some(&Value::I64(parent)))
1973                    .map(|row| (row.get("id").cloned(), row.get("name").cloned()))
1974                    .collect::<Vec<_>>()
1975            };
1976
1977            for parent in [11_i64, 12_i64] {
1978                assert_eq!(children_of(&probes, parent), children_of(&window, parent));
1979            }
1980            assert_eq!(
1981                children_of(&window, 11),
1982                vec![
1983                    (Some(Value::I64(1105)), Some(Value::Text("visible".into()))),
1984                    (Some(Value::I64(1103)), Some(Value::Text("visible".into()))),
1985                    (Some(Value::I64(1102)), Some(Value::Text("visible".into()))),
1986                ]
1987            );
1988            assert_eq!(
1989                children_of(&window, 12),
1990                vec![
1991                    (Some(Value::I64(1205)), Some(Value::Text("visible".into()))),
1992                    (Some(Value::I64(1203)), Some(Value::Text("visible".into()))),
1993                    (Some(Value::I64(1202)), Some(Value::Text("visible".into()))),
1994                ]
1995            );
1996            assert!(children_of(&probes, 13).is_empty());
1997            assert!(children_of(&window, 13).is_empty());
1998        });
1999    }
2000
2001    #[test]
2002    fn sqlite_boolean_new_schema_roundtrips_as_bool() {
2003        let executor =
2004            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
2005        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
2006        let ddl = SqliteDialect.compile_create_table(&entity).unwrap();
2007        assert!(ddl.contains("enabled BOOLEAN NOT NULL"), "{ddl}");
2008        assert!(ddl.contains("optional_enabled BOOLEAN"), "{ddl}");
2009        assert!(!ddl.contains("enabled INTEGER"), "{ddl}");
2010
2011        executor.ensure_schema(&SqliteDialect, &[&entity]).unwrap();
2012        for (id, enabled, optional_enabled) in [(1_u64, false, true), (2_u64, true, false)] {
2013            let insert = SqliteDialect
2014                .compile_insert(
2015                    &entity,
2016                    &InsertCommand::new("FeatureFlag")
2017                        .value("id", id)
2018                        .value("version", 1_i64)
2019                        .value("enabled", enabled)
2020                        .value("optional_enabled", optional_enabled),
2021                )
2022                .unwrap();
2023            assert_eq!(executor.execute(&insert).unwrap(), 1);
2024        }
2025
2026        let select = SqliteDialect
2027            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
2028            .unwrap();
2029        let rows = executor.fetch_all_compact(&select).unwrap();
2030        assert_eq!(rows[0].get("enabled"), Some(&Value::Bool(false)));
2031        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::Bool(true)));
2032        assert_eq!(rows[1].get("enabled"), Some(&Value::Bool(true)));
2033        assert_eq!(rows[1].get("optional_enabled"), Some(&Value::Bool(false)));
2034
2035        let first =
2036            <FeatureFlagRow as teaql_core::Entity>::from_compact_row(rows[0].clone()).unwrap();
2037        let second =
2038            <FeatureFlagRow as teaql_core::Entity>::from_compact_row(rows[1].clone()).unwrap();
2039        assert!(!first.enabled);
2040        assert_eq!(first.optional_enabled, Some(true));
2041        assert!(second.enabled);
2042        assert_eq!(second.optional_enabled, Some(false));
2043    }
2044
2045    #[test]
2046    fn sqlite_boolean_legacy_integer_schema_maps_only_binary_values() {
2047        let executor =
2048            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
2049        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
2050        executor
2051            .execute(&CompiledQuery {
2052                sql: "CREATE TABLE feature_flags (id INTEGER PRIMARY KEY, version INTEGER NOT NULL, enabled INTEGER NOT NULL, optional_enabled INTEGER)"
2053                    .to_owned(),
2054                params: Vec::new(),
2055                comment: None,
2056            })
2057            .unwrap();
2058
2059        let insert = SqliteDialect
2060            .compile_insert(
2061                &entity,
2062                &InsertCommand::new("FeatureFlag")
2063                    .value("id", 1_u64)
2064                    .value("version", 1_i64)
2065                    .value("enabled", true)
2066                    .value("optional_enabled", false),
2067            )
2068            .unwrap();
2069        executor.execute(&insert).unwrap();
2070        executor
2071            .execute(&CompiledQuery {
2072                sql: "INSERT INTO feature_flags (id, version, enabled, optional_enabled) VALUES (?, ?, ?, ?)"
2073                    .to_owned(),
2074                params: vec![
2075                    Value::U64(2),
2076                    Value::I64(1),
2077                    Value::I64(2),
2078                    Value::Null,
2079                ],
2080                comment: None,
2081            })
2082            .unwrap();
2083        let select = SqliteDialect
2084            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
2085            .unwrap();
2086        let rows = executor.fetch_all_compact(&select).unwrap();
2087        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
2088        assert_eq!(rows[0].get("enabled"), Some(&Value::I64(1)));
2089        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::I64(0)));
2090
2091        let decoded =
2092            <FeatureFlagRow as teaql_core::Entity>::from_compact_row(rows[0].clone()).unwrap();
2093        assert!(decoded.enabled);
2094        assert_eq!(decoded.optional_enabled, Some(false));
2095        assert_eq!(rows[1].get("enabled"), Some(&Value::I64(2)));
2096        let error =
2097            <FeatureFlagRow as teaql_core::Entity>::from_compact_row(rows[1].clone()).unwrap_err();
2098        assert!(error.message.contains("invalid field enabled"));
2099
2100        for (value, expected) in [
2101            (Value::I64(0), false),
2102            (Value::I64(1), true),
2103            (Value::U64(0), false),
2104            (Value::U64(1), true),
2105        ] {
2106            let decoded = <FeatureFlagRow as teaql_core::Entity>::from_compact_row(
2107                teaql_core::CompactRow::from_map(feature_flag_record(value, Value::Null)),
2108            )
2109            .unwrap();
2110            assert_eq!(decoded.enabled, expected);
2111            assert_eq!(decoded.optional_enabled, None);
2112        }
2113
2114        for invalid in [Value::I64(-1), Value::I64(2), Value::U64(2)] {
2115            let error = <FeatureFlagRow as teaql_core::Entity>::from_compact_row(
2116                teaql_core::CompactRow::from_map(feature_flag_record(invalid, Value::Null)),
2117            )
2118            .unwrap_err();
2119            assert!(error.message.contains("invalid field enabled"));
2120        }
2121        let error = <FeatureFlagRow as teaql_core::Entity>::from_compact_row(
2122            teaql_core::CompactRow::from_map(feature_flag_record(Value::Bool(true), Value::U64(2))),
2123        )
2124        .unwrap_err();
2125        assert!(error.message.contains("invalid field optional_enabled"));
2126    }
2127
2128    #[test]
2129    fn sqlite_executor_parses_json_only_for_json_columns() {
2130        let executor =
2131            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
2132
2133        executor
2134            .execute(&CompiledQuery {
2135                sql: "CREATE TABLE payloads (text_payload TEXT, json_payload JSON)".to_owned(),
2136                params: Vec::new(),
2137                comment: None,
2138            })
2139            .unwrap();
2140        executor
2141            .execute(&CompiledQuery {
2142                sql: "INSERT INTO payloads (text_payload, json_payload) VALUES (?, ?)".to_owned(),
2143                params: vec![
2144                    Value::Text("{\"active\":true}".to_owned()),
2145                    Value::Json(serde_json::json!({"active": true})),
2146                ],
2147                comment: None,
2148            })
2149            .unwrap();
2150
2151        let rows = executor
2152            .fetch_all_compact(&CompiledQuery {
2153                sql: "SELECT text_payload, json_payload FROM payloads".to_owned(),
2154                params: Vec::new(),
2155                comment: None,
2156            })
2157            .unwrap();
2158
2159        assert_eq!(
2160            rows[0].get("text_payload"),
2161            Some(&Value::Text("{\"active\":true}".to_owned()))
2162        );
2163        assert_eq!(
2164            rows[0].get("json_payload"),
2165            Some(&Value::Json(serde_json::json!({"active": true})))
2166        );
2167    }
2168
2169    #[test]
2170    fn sqlite_id_space_generator_increments_ids() {
2171        let executor =
2172            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
2173        let generator = SqliteIdSpaceGenerator::from_executor(executor);
2174        assert_eq!(generator.next_id("Order").unwrap(), 1);
2175        assert_eq!(generator.next_id("Order").unwrap(), 2);
2176    }
2177
2178    #[test]
2179    fn sqlite_id_space_generator_is_safe_across_connections() {
2180        let path = std::env::temp_dir().join(format!(
2181            "teaql-id-space-{}-{}.db",
2182            std::process::id(),
2183            std::time::SystemTime::now()
2184                .duration_since(std::time::UNIX_EPOCH)
2185                .unwrap()
2186                .as_nanos()
2187        ));
2188        let mut workers = Vec::new();
2189        for _ in 0..4 {
2190            let path = path.clone();
2191            workers.push(std::thread::spawn(move || {
2192                let connection = Connection::open(path).unwrap();
2193                connection
2194                    .busy_timeout(std::time::Duration::from_secs(5))
2195                    .unwrap();
2196                let generator = SqliteIdSpaceGenerator::new(connection);
2197                (0..25)
2198                    .map(|_| generator.next_id("Order").unwrap())
2199                    .collect::<Vec<_>>()
2200            }));
2201        }
2202        let mut ids = workers
2203            .into_iter()
2204            .flat_map(|worker| worker.join().unwrap())
2205            .collect::<Vec<_>>();
2206        ids.sort_unstable();
2207        assert_eq!(ids, (1..=100).collect::<Vec<_>>());
2208        let _ = std::fs::remove_file(path);
2209    }
2210
2211    #[test]
2212    fn sqlite_fetch_stream_returns_chunked_rows() {
2213        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
2214            Connection::open_in_memory().unwrap(),
2215        )));
2216        let entity = entity();
2217
2218        // Create table and insert 25 rows
2219        executor
2220            .execute(&CompiledQuery {
2221                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
2222                    .to_owned(),
2223                params: Vec::new(),
2224                comment: None,
2225            })
2226            .unwrap();
2227
2228        for i in 1..=25 {
2229            let insert = SqliteDialect
2230                .compile_insert(
2231                    &entity,
2232                    &InsertCommand::new("Order")
2233                        .value("id", i as u64)
2234                        .value("version", 1_i64)
2235                        .value("name", format!("order-{i}")),
2236                )
2237                .unwrap();
2238            executor.execute(&insert).unwrap();
2239        }
2240
2241        // Stream with chunk_size = 10
2242        let query = SelectQuery::new("Order")
2243            .filter(Expr::gt("version", 0_i64))
2244            .order_asc("id")
2245            .stream(10);
2246
2247        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
2248
2249        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
2250
2251        // 25 rows / 10 per chunk = 3 chunks
2252        assert_eq!(chunks.len(), 3);
2253        assert_eq!(chunks[0].rows.len(), 10);
2254        assert_eq!(chunks[0].chunk_index, 0);
2255        assert!(!chunks[0].is_last);
2256
2257        assert_eq!(chunks[1].rows.len(), 10);
2258        assert_eq!(chunks[1].chunk_index, 1);
2259        assert!(!chunks[1].is_last);
2260
2261        assert_eq!(chunks[2].rows.len(), 5);
2262        assert_eq!(chunks[2].chunk_index, 2);
2263        assert!(chunks[2].is_last);
2264
2265        // Verify first and last row
2266        assert_eq!(
2267            chunks[0].rows[0].get("name"),
2268            Some(&Value::Text("order-1".to_owned()))
2269        );
2270        assert_eq!(
2271            chunks[2].rows[4].get("name"),
2272            Some(&Value::Text("order-25".to_owned()))
2273        );
2274    }
2275
2276    #[test]
2277    fn sqlite_fetch_stream_handles_empty_result() {
2278        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
2279            Connection::open_in_memory().unwrap(),
2280        )));
2281
2282        executor
2283            .execute(&CompiledQuery {
2284                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
2285                    .to_owned(),
2286                params: Vec::new(),
2287                comment: None,
2288            })
2289            .unwrap();
2290
2291        let entity = entity();
2292        let query = SelectQuery::new("Order")
2293            .filter(Expr::gt("version", 0_i64))
2294            .stream(10);
2295
2296        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
2297
2298        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
2299
2300        // Empty result = 1 chunk with 0 rows, marked as last
2301        assert_eq!(chunks.len(), 1);
2302        assert_eq!(chunks[0].rows.len(), 0);
2303        assert!(chunks[0].is_last);
2304    }
2305
2306    #[test]
2307    fn sqlite_fetch_stream_exact_chunk_boundary() {
2308        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
2309            Connection::open_in_memory().unwrap(),
2310        )));
2311        let entity = entity();
2312
2313        executor
2314            .execute(&CompiledQuery {
2315                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
2316                    .to_owned(),
2317                params: Vec::new(),
2318                comment: None,
2319            })
2320            .unwrap();
2321
2322        // Insert exactly 20 rows
2323        for i in 1..=20 {
2324            let insert = SqliteDialect
2325                .compile_insert(
2326                    &entity,
2327                    &InsertCommand::new("Order")
2328                        .value("id", i as u64)
2329                        .value("version", 1_i64)
2330                        .value("name", format!("order-{i}")),
2331                )
2332                .unwrap();
2333            executor.execute(&insert).unwrap();
2334        }
2335
2336        let query = SelectQuery::new("Order")
2337            .filter(Expr::gt("version", 0_i64))
2338            .order_asc("id")
2339            .stream(10);
2340
2341        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
2342
2343        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
2344
2345        // 20 rows / 10 per chunk = 2 full chunks + 1 empty final chunk
2346        assert_eq!(chunks.len(), 3);
2347        assert_eq!(chunks[0].rows.len(), 10);
2348        assert!(!chunks[0].is_last);
2349        assert_eq!(chunks[1].rows.len(), 10);
2350        assert!(!chunks[1].is_last);
2351        assert_eq!(chunks[2].rows.len(), 0);
2352        assert!(chunks[2].is_last);
2353    }
2354
2355    #[test]
2356    fn test_parse_sqlite_timestamp() {
2357        let ts1 = parse_sqlite_timestamp("2023-01-01 12:30:45").unwrap();
2358        assert!(matches!(ts1, Value::Timestamp(_)));
2359
2360        let ts2 = parse_sqlite_timestamp("2023-01-01").unwrap();
2361        assert!(matches!(ts2, Value::Timestamp(_)));
2362
2363        let ts3 = parse_sqlite_timestamp("2023-01-01T12:30:45Z").unwrap();
2364        assert!(matches!(ts3, Value::Timestamp(_)));
2365
2366        let ts4 = parse_sqlite_timestamp("2026-08-23 10:43:16.152546+00").unwrap();
2367        assert!(matches!(ts4, Value::Timestamp(_)));
2368
2369        let ts5 = parse_sqlite_timestamp("2026-08-23 10:43:16.152546").unwrap();
2370        assert!(matches!(ts5, Value::Timestamp(_)));
2371
2372        assert_eq!(
2373            parse_fixed_sqlite_timestamp("2024-01-01 00:00:00+00"),
2374            Some(1_704_067_200_000)
2375        );
2376        assert_eq!(
2377            parse_fixed_sqlite_timestamp("2024-01-01T08:00:00.123+08:00"),
2378            Some(1_704_067_200_123)
2379        );
2380        assert_eq!(
2381            parse_fixed_sqlite_timestamp("2023-12-31 19:00:00-0500"),
2382            Some(1_704_067_200_000)
2383        );
2384        assert_eq!(parse_fixed_sqlite_timestamp("2024-13-01 00:00:00Z"), None);
2385        assert_eq!(parse_fixed_sqlite_timestamp("2024-01-01 00:00:00+24"), None);
2386
2387        assert!(parse_sqlite_timestamp("invalid").is_err());
2388    }
2389
2390    #[test]
2391    fn declared_text_does_not_infer_timestamp_from_content() {
2392        for decl_type in ["TEXT", "VARCHAR(255)", "CHAR(32)", "CLOB"] {
2393            let column = ColumnInfo {
2394                name: "external_timestamp".to_owned(),
2395                decode_kind: sqlite_decode_kind(Some(decl_type)),
2396            };
2397
2398            assert_eq!(
2399                decode_sqlite_text(b"2024-01-01 00:57:55", &column).unwrap(),
2400                Value::Text("2024-01-01 00:57:55".to_owned())
2401            );
2402        }
2403    }
2404
2405    #[test]
2406    fn declared_column_types_compile_to_decode_kinds() {
2407        assert_eq!(sqlite_decode_kind(Some("BOOLEAN")), SqliteDecodeKind::Bool);
2408        assert_eq!(
2409            sqlite_decode_kind(Some("decimal(20, 4)")),
2410            SqliteDecodeKind::Decimal
2411        );
2412        assert_eq!(
2413            sqlite_decode_kind(Some(" VARCHAR(255) ")),
2414            SqliteDecodeKind::Text
2415        );
2416        assert_eq!(
2417            sqlite_decode_kind(Some("datetime")),
2418            SqliteDecodeKind::Timestamp
2419        );
2420        assert_eq!(sqlite_decode_kind(Some("custom")), SqliteDecodeKind::Infer);
2421        assert_eq!(sqlite_decode_kind(None), SqliteDecodeKind::Infer);
2422    }
2423}
teaql_provider_sqlite/lib.rs

teaql_provider_sqlite/
lib.rs
