teaql_provider_sqlite/
lib.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::future::Future;
3use std::pin::Pin;
4use std::str::FromStr;
5use std::sync::{Arc, Mutex, MutexGuard};
6
7use chrono::{DateTime, NaiveDate, NaiveDateTime};
8use rusqlite::types::{Value as SqliteValue, ValueRef};
9use rusqlite::{Connection, Row, params_from_iter};
10use rust_decimal::Decimal;
11use teaql_core::{
12    DataType, EntityDescriptor, Expr, InsertCommand, PropertyDescriptor, Record, SelectQuery,
13    UpdateCommand, Value,
14};
15use teaql_runtime::{
16    GraphNode, InternalIdGenerator, RawAuditEvent, RuntimeError, SchemaProvider, UserContext,
17};
18use teaql_sql::{
19    CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, SqlTransport,
20    quote_identifier_if_needed,
21};
22
23pub const DEFAULT_ID_SPACE_TABLE: &str = "teaql_id_space";
24
25#[derive(Debug, Default, Clone, Copy)]
26pub struct SqliteDialect;
27
28impl SqlDialect for SqliteDialect {
29    fn kind(&self) -> DatabaseKind {
30        DatabaseKind::Sqlite
31    }
32
33    fn quote_ident(&self, ident: &str) -> String {
34        quote_ident(ident)
35    }
36
37    fn placeholder(&self, _index: usize) -> String {
38        "?".to_owned()
39    }
40
41    fn schema_type_sql(
42        &self,
43        data_type: DataType,
44        property: &PropertyDescriptor,
45    ) -> Result<&'static str, SqlCompileError> {
46        match data_type {
47            DataType::Bool => Ok("BOOLEAN"),
48            DataType::I64 | DataType::U64 if property.is_id => Ok("INTEGER"),
49            DataType::I64 | DataType::U64 => Ok("INTEGER"),
50            DataType::F64 => Ok("REAL"),
51            DataType::Decimal => Ok("NUMERIC"),
52            DataType::Text => Ok("VARCHAR(255)"),
53            DataType::LargeText => Ok("TEXT"),
54            DataType::Json => Ok("JSON"),
55            DataType::Date => Ok("DATE"),
56            DataType::Timestamp => Ok("TIMESTAMP"),
57        }
58    }
59
60    fn compile_add_column(
61        &self,
62        entity: &EntityDescriptor,
63        property: &PropertyDescriptor,
64    ) -> Result<String, SqlCompileError> {
65        // SQLite does not support adding NOT NULL columns without a DEFAULT.
66        // Since TeaQL enforces nullability at the application layer, we can safely
67        // strip the NOT NULL constraint when adding columns to existing tables.
68        let def = self.column_definition_sql(property)?;
69        let def_without_not_null = def.replace(" NOT NULL", "");
70
71        Ok(format!(
72            "ALTER TABLE {} ADD COLUMN {}",
73            self.quote_ident(&entity.table_name),
74            def_without_not_null
75        ))
76    }
77}
78
79#[derive(Debug)]
80pub enum MutationExecutorError {
81    Sqlite(rusqlite::Error),
82    SqlCompile(SqlCompileError),
83    UnsupportedValue(&'static str),
84    UnsupportedColumnType(String),
85    Bind(String),
86    Lock(String),
87}
88
89impl std::fmt::Display for MutationExecutorError {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        match self {
92            Self::Sqlite(err) => err.fmt(f),
93            Self::SqlCompile(err) => err.fmt(f),
94            Self::UnsupportedValue(kind) => {
95                write!(
96                    f,
97                    "unsupported rusqlite bind value for mutation executor: {kind}"
98                )
99            }
100            Self::UnsupportedColumnType(kind) => {
101                write!(
102                    f,
103                    "unsupported rusqlite column type for record decoding: {kind}"
104                )
105            }
106            Self::Bind(message) => write!(f, "rusqlite bind error: {message}"),
107            Self::Lock(message) => write!(f, "rusqlite connection lock error: {message}"),
108        }
109    }
110}
111
112impl std::error::Error for MutationExecutorError {}
113
114impl From<rusqlite::Error> for MutationExecutorError {
115    fn from(value: rusqlite::Error) -> Self {
116        Self::Sqlite(value)
117    }
118}
119
120impl From<SqlCompileError> for MutationExecutorError {
121    fn from(value: SqlCompileError) -> Self {
122        Self::SqlCompile(value)
123    }
124}
125
126#[derive(Clone)]
127pub struct SqliteMutationExecutor {
128    connection: Arc<Mutex<Connection>>,
129}
130
131impl SqliteMutationExecutor {
132    pub fn new(connection: Arc<Mutex<Connection>>) -> Self {
133        Self { connection }
134    }
135
136    pub fn from_connection(connection: Connection) -> Self {
137        Self::new(Arc::new(Mutex::new(connection)))
138    }
139
140    pub fn connection(&self) -> Arc<Mutex<Connection>> {
141        Arc::clone(&self.connection)
142    }
143
144    pub fn ensure_schema(
145        &self,
146        dialect: &SqliteDialect,
147        entities: &[&EntityDescriptor],
148    ) -> Result<(), MutationExecutorError> {
149        self.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
150
151        for entity in entities {
152            if !self.table_exists(&entity.table_name)? {
153                let sql = dialect.compile_create_table(entity)?;
154                self.lock()?.execute(&sql, [])?;
155                continue;
156            }
157
158            let existing_columns = self.table_columns(&entity.table_name)?;
159            for property in &entity.properties {
160                let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
161                if existing_columns.contains(&bare_column) {
162                    continue;
163                }
164                let sql = dialect.compile_add_column(entity, property)?;
165                self.lock()?.execute(&sql, [])?;
166            }
167
168            for sql in dialect.schema_indexes_sqls(entity)? {
169                self.lock()?.execute(&sql, [])?;
170            }
171        }
172        Ok(())
173    }
174
175    pub fn ensure_id_space_table(&self, table_name: &str) -> Result<(), MutationExecutorError> {
176        let sql = format!(
177            "CREATE TABLE IF NOT EXISTS {} (type_name VARCHAR(100) PRIMARY KEY, current_level BIGINT NOT NULL)",
178            quote_ident(table_name)
179        );
180        self.lock()?.execute(&sql, [])?;
181        Ok(())
182    }
183
184    pub fn begin_transaction(&self) -> Result<(), MutationExecutorError> {
185        self.lock()?.execute("BEGIN IMMEDIATE", [])?;
186        Ok(())
187    }
188
189    pub fn commit_transaction(&self) -> Result<(), MutationExecutorError> {
190        self.lock()?.execute("COMMIT", [])?;
191        Ok(())
192    }
193
194    pub fn rollback_transaction(&self) -> Result<(), MutationExecutorError> {
195        self.lock()?.execute("ROLLBACK", [])?;
196        Ok(())
197    }
198
199    pub fn execute(&self, query: &CompiledQuery) -> Result<u64, MutationExecutorError> {
200        let params = bind_values(&query.params)?;
201        let rows = self
202            .lock()?
203            .execute(&query.sql_with_comment(), params_from_iter(params.iter()))?;
204        Ok(rows as u64)
205    }
206
207    pub fn fetch_all(&self, query: &CompiledQuery) -> Result<Vec<Record>, MutationExecutorError> {
208        let params = bind_values(&query.params)?;
209        let connection = self.lock()?;
210        let mut statement = connection.prepare(&query.sql_with_comment())?;
211        let columns = statement_columns(&statement);
212        let mut rows = statement.query(params_from_iter(params.iter()))?;
213        let mut records = Vec::new();
214        while let Some(row) = rows.next()? {
215            records.push(decode_sqlite_row(row, &columns)?);
216        }
217        Ok(records)
218    }
219
220    /// Fetch rows in streaming mode (chunked).
221    /// Returns a Vec of StreamChunk, each containing up to `chunk_size` rows.
222    pub fn fetch_stream(
223        &self,
224        query: &CompiledQuery,
225        chunk_size: usize,
226    ) -> Result<Vec<teaql_data_service::StreamChunk>, MutationExecutorError> {
227        let params = bind_values(&query.params)?;
228        let connection = self.lock()?;
229        let mut statement = connection.prepare(&query.sql_with_comment())?;
230        let columns = statement_columns(&statement);
231        let mut rows = statement.query(params_from_iter(params.iter()))?;
232
233        let mut chunks = Vec::new();
234        let mut current_chunk = Vec::new();
235        let mut chunk_index = 0;
236
237        while let Some(row) = rows.next()? {
238            current_chunk.push(decode_sqlite_row(row, &columns)?);
239            if current_chunk.len() >= chunk_size {
240                chunks.push(teaql_data_service::StreamChunk {
241                    rows: current_chunk,
242                    chunk_index,
243                    is_last: false,
244                });
245                current_chunk = Vec::new();
246                chunk_index += 1;
247            }
248        }
249
250        // Push the final chunk (may be empty if exactly aligned)
251        chunks.push(teaql_data_service::StreamChunk {
252            rows: current_chunk,
253            chunk_index,
254            is_last: true,
255        });
256
257        Ok(chunks)
258    }
259
260    pub fn table_exists(&self, table_name: &str) -> Result<bool, MutationExecutorError> {
261        let exists: i64 = self.lock()?.query_row(
262            "SELECT COUNT(1) FROM sqlite_master WHERE type = 'table' AND name = ?",
263            [table_name],
264            |row| row.get(0),
265        )?;
266        Ok(exists > 0)
267    }
268
269    pub fn table_columns(
270        &self,
271        table_name: &str,
272    ) -> Result<BTreeSet<String>, MutationExecutorError> {
273        let pragma_sql = format!("PRAGMA table_info({})", quote_ident(table_name));
274        let connection = self.lock()?;
275        let mut statement = connection.prepare(&pragma_sql)?;
276        let rows = statement.query_map([], |row| row.get::<_, String>("name"))?;
277        let mut columns = BTreeSet::new();
278        for row in rows {
279            columns.insert(row?.to_lowercase());
280        }
281        Ok(columns)
282    }
283
284    fn lock(&self) -> Result<MutexGuard<'_, Connection>, MutationExecutorError> {
285        self.connection
286            .lock()
287            .map_err(|err| MutationExecutorError::Lock(err.to_string()))
288    }
289}
290
291impl teaql_data_service::DataServiceExecutor for SqliteMutationExecutor {
292    type Error = MutationExecutorError;
293
294    fn capabilities(&self) -> teaql_data_service::DataServiceCapabilities {
295        teaql_data_service::DataServiceCapabilities {
296            query: true,
297            mutation: true,
298            transaction: true,
299            schema: true,
300            id_generation: true,
301            ..Default::default()
302        }
303    }
304}
305
306impl SqlTransport for SqliteMutationExecutor {
307    type Error = MutationExecutorError;
308
309    async fn fetch_all_sql(&self, query: &CompiledQuery) -> Result<Vec<Record>, Self::Error> {
310        SqliteMutationExecutor::fetch_all(self, query)
311    }
312
313    async fn execute_sql(&self, query: &CompiledQuery) -> Result<u64, Self::Error> {
314        SqliteMutationExecutor::execute(self, query)
315    }
316}
317
318impl teaql_data_service::StreamQueryExecutor for SqliteMutationExecutor {
319    async fn query_stream(
320        &self,
321        request: teaql_data_service::QueryRequest,
322        chunk_size: usize,
323    ) -> Result<Vec<teaql_data_service::StreamChunk>, Self::Error> {
324        let dialect = SqliteDialect;
325        // Use a dummy entity descriptor for compilation
326        let entity_desc = teaql_core::EntityDescriptor::new(&request.query.entity);
327        let compiled = dialect
328            .compile_select(&entity_desc, &request.query)
329            .map_err(MutationExecutorError::SqlCompile)?;
330        SqliteMutationExecutor::fetch_stream(self, &compiled, chunk_size)
331    }
332}
333
334impl teaql_sql::SqlTransaction for SqliteMutationExecutor {
335    type Error = MutationExecutorError;
336
337    async fn commit_sql(self) -> Result<(), Self::Error> {
338        self.commit_transaction()
339    }
340
341    async fn rollback_sql(self) -> Result<(), Self::Error> {
342        self.rollback_transaction()
343    }
344}
345
346impl teaql_sql::SqlTransactionTransport for SqliteMutationExecutor {
347    type Tx<'a>
348        = Self
349    where
350        Self: 'a;
351
352    async fn begin_sql(&self) -> Result<Self::Tx<'_>, Self::Error> {
353        self.begin_transaction()?;
354        Ok(self.clone())
355    }
356}
357
358fn initial_graph_exists_sqlite(
359    executor: &SqliteMutationExecutor,
360    dialect: &SqliteDialect,
361    entity: &EntityDescriptor,
362    graph: &GraphNode,
363) -> Result<bool, MutationExecutorError> {
364    let Some(id) = graph.values.get("id") else {
365        return Ok(false);
366    };
367    let query = dialect.compile_select(
368        entity,
369        &SelectQuery::new(&graph.entity)
370            .project("id")
371            .filter(Expr::eq("id", id.clone()))
372            .limit(1),
373    )?;
374    Ok(!executor.fetch_all(&query)?.is_empty())
375}
376
377fn compile_initial_graph_insert(
378    dialect: &impl SqlDialect,
379    entity: &EntityDescriptor,
380    graph: &GraphNode,
381) -> Result<CompiledQuery, MutationExecutorError> {
382    let mut command = InsertCommand::new(&graph.entity);
383    for (field, value) in &graph.values {
384        command = command.value(field.clone(), value.clone());
385    }
386    dialect.compile_insert(entity, &command).map_err(Into::into)
387}
388
389fn compile_initial_graph_update(
390    dialect: &impl SqlDialect,
391    entity: &EntityDescriptor,
392    graph: &GraphNode,
393) -> Result<Option<CompiledQuery>, MutationExecutorError> {
394    let Some(id) = graph.values.get("id") else {
395        return Ok(None);
396    };
397    let mut command = UpdateCommand::new(&graph.entity, id.clone());
398    for (field, value) in &graph.values {
399        if field == "id" {
400            continue;
401        }
402        command = command.value(field.clone(), value.clone());
403    }
404    match dialect.compile_update(entity, &command) {
405        Ok(query) => Ok(Some(query)),
406        Err(SqlCompileError::EmptyMutation(_)) => Ok(None),
407        Err(err) => Err(err.into()),
408    }
409}
410
411pub trait SqliteSchemaExt {
412    fn ensure_sqlite_schema(
413        &self,
414    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + Send + '_>>;
415}
416
417pub fn ensure_sqlite_schema_for(ctx: &UserContext) -> Result<(), MutationExecutorError> {
418    let dialect = ctx.get_resource::<SqliteDialect>().ok_or_else(|| {
419        MutationExecutorError::Bind("missing typed resource: SqliteDialect".to_owned())
420    })?;
421    let executor = ctx
422        .get_resource::<SqliteMutationExecutor>()
423        .ok_or_else(|| {
424            MutationExecutorError::Bind("missing typed resource: SqliteMutationExecutor".to_owned())
425        })?;
426
427    let entities = ctx.all_entities();
428
429    // Ensure id space table exists
430    executor.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
431
432    // Process each entity table individually with granular events
433    for entity in &entities {
434        let field_count = entity.properties.len();
435        if !executor.table_exists(&entity.table_name)? {
436            // New table: create it
437            let sql = dialect.compile_create_table(entity)?;
438            executor.lock()?.execute(&sql, [])?;
439            let _ = ctx.send_event(RawAuditEvent::schema_created(
440                &entity.name,
441                &entity.table_name,
442                field_count,
443            ));
444            continue;
445        }
446        // Existing table: check for missing columns
447        let existing_columns = executor.table_columns(&entity.table_name)?;
448        let mut fields_added = 0;
449        for property in &entity.properties {
450            let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
451            if existing_columns.contains(&bare_column) {
452                continue;
453            }
454            let sql = dialect.compile_add_column(entity, property)?;
455            executor.lock()?.execute(&sql, [])?;
456            let _ = ctx.send_event(RawAuditEvent::field_added(
457                &entity.name,
458                &entity.table_name,
459                &property.column_name,
460            ));
461            fields_added += 1;
462        }
463        let _ = ctx.send_event(RawAuditEvent::schema_verified(
464            &entity.name,
465            &entity.table_name,
466            field_count,
467        ));
468        let _ = fields_added; // used above for FieldAdded events
469    }
470
471    // Seed initial data, tracking insert vs update counts per entity
472    let mut seed_counts: BTreeMap<String, (usize, usize)> = BTreeMap::new(); // (inserted, updated)
473    for graph in ctx.initial_graphs() {
474        let entity = ctx.entity(&graph.entity).ok_or_else(|| {
475            MutationExecutorError::Bind(format!("missing entity: {}", graph.entity))
476        })?;
477        let counts = seed_counts.entry(graph.entity.clone()).or_insert((0, 0));
478        if initial_graph_exists_sqlite(executor, dialect, entity, graph)? {
479            if let Some(query) = compile_initial_graph_update(dialect, entity, graph)? {
480                executor.execute(&query)?;
481            }
482            counts.1 += 1; // updated
483            continue;
484        }
485        let query = compile_initial_graph_insert(dialect, entity, graph)?;
486        executor.execute(&query)?;
487        counts.0 += 1; // inserted
488    }
489
490    // Fire DataSeeded events per entity type
491    for (entity_name, (inserted, updated)) in &seed_counts {
492        let entity = ctx.entity(entity_name).ok_or_else(|| {
493            MutationExecutorError::Bind(format!("missing entity: {}", entity_name))
494        })?;
495        let _ = ctx.send_event(RawAuditEvent::data_seeded(
496            entity_name,
497            &entity.table_name,
498            *inserted,
499            *updated,
500        ));
501    }
502
503    Ok(())
504}
505
506impl SqliteSchemaExt for UserContext {
507    fn ensure_sqlite_schema(
508        &self,
509    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + Send + '_>> {
510        Box::pin(async move { ensure_sqlite_schema_for(self) })
511    }
512}
513
514#[derive(Debug, Default, Clone, Copy)]
515pub struct SqliteSchemaProvider;
516
517impl SchemaProvider for SqliteSchemaProvider {
518    fn ensure_schema<'a>(
519        &'a self,
520        ctx: &'a UserContext,
521    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>> {
522        Box::pin(async move {
523            ensure_sqlite_schema_for(ctx).map_err(|err| RuntimeError::Schema(err.to_string()))
524        })
525    }
526}
527
528pub trait SqliteProviderExt {
529    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self;
530}
531
532impl SqliteProviderExt for UserContext {
533    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self {
534        self.insert_resource(SqliteDialect);
535        self.insert_resource(executor);
536        self.set_schema_provider(SqliteSchemaProvider);
537        self
538    }
539}
540
541#[derive(Clone)]
542pub struct SqliteIdSpaceGenerator {
543    executor: SqliteMutationExecutor,
544    table_name: String,
545}
546
547impl SqliteIdSpaceGenerator {
548    pub fn new(connection: Connection) -> Self {
549        Self::from_executor(SqliteMutationExecutor::from_connection(connection))
550    }
551
552    pub fn from_executor(executor: SqliteMutationExecutor) -> Self {
553        Self {
554            executor,
555            table_name: DEFAULT_ID_SPACE_TABLE.to_owned(),
556        }
557    }
558
559    pub fn with_table_name(mut self, table_name: impl Into<String>) -> Self {
560        self.table_name = table_name.into();
561        self
562    }
563
564    pub fn ensure_table(&self) -> Result<(), MutationExecutorError> {
565        self.executor.ensure_id_space_table(&self.table_name)
566    }
567
568    pub fn next_id(&self, entity: &str) -> Result<u64, MutationExecutorError> {
569        self.ensure_table()?;
570        let sql = format!(
571            "INSERT INTO {} (type_name, current_level) VALUES (?, 1) \
572             ON CONFLICT (type_name) DO UPDATE \
573             SET current_level = current_level + 1 \
574             RETURNING current_level",
575            quote_ident(&self.table_name)
576        );
577        let id: i64 = self
578            .executor
579            .lock()?
580            .query_row(&sql, [entity], |row| row.get(0))?;
581        u64::try_from(id).map_err(|_| {
582            MutationExecutorError::Bind(format!("generated id {id} cannot be represented as u64"))
583        })
584    }
585}
586
587impl InternalIdGenerator for SqliteIdSpaceGenerator {
588    fn generate_id(&self, entity: &str) -> Result<u64, RuntimeError> {
589        self.next_id(entity)
590            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))
591    }
592}
593
594fn quote_ident(ident: &str) -> String {
595    quote_identifier_if_needed(ident, '"')
596}
597
598/// Strip wrapping identifier quotes from a SQL identifier.
599///
600/// SQLite `PRAGMA table_info` returns bare column names (e.g. `description`),
601/// but generated `PropertyDescriptor::column_name` may carry quotes
602/// (e.g. `"description"`) when the name is a reserved keyword.  This helper
603/// normalises the column name so the two can be compared correctly during
604/// schema migration.
605fn strip_identifier_quotes(ident: &str) -> &str {
606    let bytes = ident.as_bytes();
607    if bytes.len() >= 2 {
608        let (first, last) = (bytes[0], bytes[bytes.len() - 1]);
609        if (first == b'"' && last == b'"')
610            || (first == b'`' && last == b'`')
611            || (first == b'[' && last == b']')
612        {
613            return &ident[1..ident.len() - 1];
614        }
615    }
616    ident
617}
618
619fn bind_values(values: &[Value]) -> Result<Vec<SqliteValue>, MutationExecutorError> {
620    values.iter().map(bind_sqlite_value).collect()
621}
622
623fn bind_sqlite_value(value: &Value) -> Result<SqliteValue, MutationExecutorError> {
624    match value {
625        Value::Null => Ok(SqliteValue::Null),
626        Value::Bool(v) => Ok(SqliteValue::Integer(i64::from(*v))),
627        Value::I64(v) => Ok(SqliteValue::Integer(*v)),
628        Value::U64(v) => i64::try_from(*v)
629            .map(SqliteValue::Integer)
630            .map_err(|_| MutationExecutorError::Bind(format!("u64 value {v} exceeds i64 range"))),
631        Value::F64(v) => Ok(SqliteValue::Real(*v)),
632        // Bind the canonical numeric spelling. SQLite NUMERIC affinity keeps
633        // predicates and aggregates numeric; an application-only text prefix
634        // makes range comparisons silently return the wrong result.
635        Value::Decimal(v) => Ok(SqliteValue::Text(v.to_string())),
636        Value::Text(v) => Ok(SqliteValue::Text(v.clone())),
637        Value::Json(v) => Ok(SqliteValue::Text(v.to_string())),
638        Value::Date(v) => Ok(SqliteValue::Text(v.format("%Y-%m-%d").to_string())),
639        Value::Timestamp(v) => Ok(SqliteValue::Text(v.0.to_string())),
640        Value::Object(_) => Err(MutationExecutorError::UnsupportedValue("object")),
641        Value::List(_) => Err(MutationExecutorError::UnsupportedValue("list")),
642        Value::TypedNull(_) => Ok(SqliteValue::Null),
643    }
644}
645
646#[derive(Debug, Clone)]
647struct ColumnInfo {
648    name: String,
649    decl_type: Option<String>,
650}
651
652fn statement_columns(statement: &rusqlite::Statement<'_>) -> Vec<ColumnInfo> {
653    statement
654        .columns()
655        .into_iter()
656        .map(|column| ColumnInfo {
657            name: column.name().to_owned(),
658            decl_type: column.decl_type().map(|value| value.to_ascii_uppercase()),
659        })
660        .collect()
661}
662
663fn decode_sqlite_row(
664    row: &Row<'_>,
665    columns: &[ColumnInfo],
666) -> Result<Record, MutationExecutorError> {
667    let mut record = BTreeMap::new();
668    for (index, column) in columns.iter().enumerate() {
669        let value_ref = row.get_ref(index)?;
670        let value = match value_ref {
671            ValueRef::Null => Value::Null,
672            ValueRef::Integer(value) => decode_sqlite_integer(value, column),
673            ValueRef::Real(value) => Value::F64(value),
674            ValueRef::Text(value) => decode_sqlite_text(value, column)?,
675            ValueRef::Blob(_) => {
676                return Err(MutationExecutorError::UnsupportedColumnType(
677                    "BLOB".to_owned(),
678                ));
679            }
680        };
681        record.insert(column.name.clone(), value);
682    }
683    Ok(record)
684}
685
686fn decode_sqlite_integer(value: i64, column: &ColumnInfo) -> Value {
687    match column_decl_type(column).as_deref() {
688        Some("BOOLEAN") | Some("BOOL") => Value::Bool(value != 0),
689        _ => Value::I64(value),
690    }
691}
692
693fn decode_sqlite_text(value: &[u8], column: &ColumnInfo) -> Result<Value, MutationExecutorError> {
694    let value = std::str::from_utf8(value)
695        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite text: {err}")))?;
696    match column_decl_type(column).as_deref() {
697        Some("NUMERIC") | Some("DECIMAL") => Decimal::from_str(value)
698            .map(Value::Decimal)
699            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite decimal: {err}"))),
700        Some("JSON") => serde_json::from_str(value).map(Value::Json).map_err(|err| {
701            MutationExecutorError::Bind(format!("invalid sqlite json value: {err}"))
702        }),
703        Some("DATE") => NaiveDate::parse_from_str(value, "%Y-%m-%d")
704            .map(Value::Date)
705            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite date: {err}"))),
706        Some("TIMESTAMP") | Some("DATETIME") => parse_sqlite_timestamp(value),
707        _ => infer_sqlite_text(value),
708    }
709}
710
711fn infer_sqlite_text(value: &str) -> Result<Value, MutationExecutorError> {
712    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
713        return Ok(Value::Date(date));
714    }
715    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
716        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
717            timestamp.timestamp_millis(),
718        )));
719    }
720    if let Ok(timestamp) = NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S") {
721        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
722            timestamp.and_utc().timestamp_millis(),
723        )));
724    }
725    Ok(Value::Text(value.to_owned()))
726}
727
728fn parse_sqlite_timestamp(value: &str) -> Result<Value, MutationExecutorError> {
729    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
730        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
731            timestamp.timestamp_millis(),
732        )));
733    }
734    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
735        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
736            date.and_hms_opt(0, 0, 0)
737                .unwrap_or_default()
738                .and_utc()
739                .timestamp_millis(),
740        )));
741    }
742    NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S")
743        .map(|timestamp| {
744            Value::Timestamp(teaql_core::time::Timestamp(
745                timestamp.and_utc().timestamp_millis(),
746            ))
747        })
748        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite timestamp: {err}")))
749}
750
751fn column_decl_type(column: &ColumnInfo) -> Option<String> {
752    column
753        .decl_type
754        .as_ref()
755        .map(|value| value.split('(').next().unwrap_or(value).trim().to_owned())
756}
757
758#[cfg(test)]
759mod tests {
760    use super::*;
761    use teaql_core::{DeleteCommand, RecoverCommand};
762    use teaql_macros::TeaqlEntity;
763    use teaql_runtime::InMemoryMetadataStore;
764
765    #[test]
766    fn decimal_bind_is_numeric_and_comparable() {
767        let value =
768            bind_sqlite_value(&Value::Decimal(Decimal::from_str("123.450").unwrap())).unwrap();
769        assert_eq!(value, SqliteValue::Text("123.450".to_owned()));
770        let connection = Connection::open_in_memory().unwrap();
771        let matches: i64 = connection
772            .query_row(
773                "SELECT 1 WHERE CAST(? AS NUMERIC) BETWEEN 120 AND 130",
774                [value],
775                |row| row.get(0),
776            )
777            .unwrap();
778        assert_eq!(matches, 1);
779    }
780
781    fn entity() -> EntityDescriptor {
782        EntityDescriptor::new("Order")
783            .table_name("orders")
784            .property(
785                PropertyDescriptor::new("id", DataType::U64)
786                    .column_name("id")
787                    .id()
788                    .not_null(),
789            )
790            .property(
791                PropertyDescriptor::new("version", DataType::I64)
792                    .column_name("version")
793                    .version()
794                    .not_null(),
795            )
796            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
797    }
798
799    fn order_line_entity() -> EntityDescriptor {
800        EntityDescriptor::new("OrderLine")
801            .table_name("order_line")
802            .property(
803                PropertyDescriptor::new("id", DataType::U64)
804                    .column_name("id")
805                    .id()
806                    .not_null(),
807            )
808            .property(
809                PropertyDescriptor::new("order_id", DataType::U64)
810                    .column_name("order_id")
811                    .not_null(),
812            )
813            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
814    }
815
816    #[allow(dead_code)]
817    #[derive(Debug, PartialEq, TeaqlEntity)]
818    #[teaql(entity = "FeatureFlag", table = "feature_flags")]
819    struct FeatureFlagRow {
820        #[teaql(id)]
821        id: u64,
822        #[teaql(version)]
823        version: i64,
824        enabled: bool,
825        optional_enabled: Option<bool>,
826    }
827
828    fn feature_flag_record(enabled: Value, optional_enabled: Value) -> Record {
829        Record::from([
830            ("id".to_owned(), Value::U64(1)),
831            ("version".to_owned(), Value::I64(1)),
832            ("enabled".to_owned(), enabled),
833            ("optional_enabled".to_owned(), optional_enabled),
834        ])
835    }
836
837    #[test]
838    fn sqlite_dialect_compiles_mutations_and_schema() {
839        let insert = SqliteDialect
840            .compile_insert(
841                &entity(),
842                &InsertCommand::new("Order")
843                    .value("id", 1_u64)
844                    .value("name", "A"),
845            )
846            .unwrap();
847        assert_eq!(insert.sql, "INSERT INTO orders (id, name) VALUES (?, ?)");
848
849        let update = SqliteDialect
850            .compile_update(
851                &entity(),
852                &UpdateCommand::new("Order", 1_u64)
853                    .expected_version(3)
854                    .value("name", "B"),
855            )
856            .unwrap();
857        assert_eq!(
858            update.sql,
859            "UPDATE orders SET name = ?, version = ? WHERE id = ? AND version = ?"
860        );
861
862        let delete = SqliteDialect
863            .compile_delete(
864                &entity(),
865                &DeleteCommand::new("Order", 1_u64).expected_version(3),
866            )
867            .unwrap();
868        let recover = SqliteDialect
869            .compile_recover(&entity(), &RecoverCommand::new("Order", 1_u64, -4))
870            .unwrap();
871        assert_eq!(
872            delete.sql,
873            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
874        );
875        assert_eq!(
876            recover.sql,
877            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
878        );
879
880        let create = SqliteDialect.compile_create_table(&entity()).unwrap();
881        assert_eq!(
882            create,
883            "CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY NOT NULL, version INTEGER NOT NULL, name VARCHAR(255))"
884        );
885    }
886
887    #[test]
888    fn sqlite_executor_ensures_schema_and_roundtrips_rows() {
889        let executor =
890            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
891        let entity = entity();
892        let mut ctx = UserContext::new()
893            .with_metadata(InMemoryMetadataStore::new().with_entity(entity.clone()));
894
895        ctx.use_sqlite_provider(executor.clone());
896        ensure_sqlite_schema_for(&ctx).unwrap();
897
898        let insert = SqliteDialect
899            .compile_insert(
900                &entity,
901                &InsertCommand::new("Order")
902                    .value("id", 1_u64)
903                    .value("version", 1_i64)
904                    .value("name", "draft"),
905            )
906            .unwrap();
907        assert_eq!(executor.execute(&insert).unwrap(), 1);
908
909        let select = SqliteDialect
910            .compile_select(
911                &entity,
912                &SelectQuery::new("Order")
913                    .filter(Expr::eq("id", 1_u64))
914                    .order_asc("id"),
915            )
916            .unwrap();
917        let rows = executor.fetch_all(&select).unwrap();
918        assert_eq!(rows.len(), 1);
919        assert_eq!(rows[0].get("id"), Some(&Value::I64(1)));
920        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
921        assert_eq!(rows[0].get("name"), Some(&Value::Text("draft".to_owned())));
922    }
923
924    #[test]
925    fn sqlite_executes_partitioned_relation_limit_per_parent() {
926        let executor =
927            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
928        let entity = order_line_entity();
929        executor.ensure_schema(&SqliteDialect, &[&entity]).unwrap();
930
931        for order_id in [11_u64, 12_u64] {
932            for index in 1_u64..=5 {
933                let id = order_id * 100 + index;
934                let insert = SqliteDialect
935                    .compile_insert(
936                        &entity,
937                        &InsertCommand::new("OrderLine")
938                            .value("id", id)
939                            .value("order_id", order_id)
940                            .value("name", format!("line-{id}")),
941                    )
942                    .unwrap();
943                executor.execute(&insert).unwrap();
944            }
945        }
946
947        let query = SelectQuery::new("OrderLine")
948            .project("id")
949            .project("order_id")
950            .order_desc("id")
951            .limit(3)
952            .partition_by("order_id");
953        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
954        let rows = executor.fetch_all(&compiled).unwrap();
955
956        assert_eq!(rows.len(), 6);
957        for order_id in [11_i64, 12_i64] {
958            let ids = rows
959                .iter()
960                .filter(|row| row.get("order_id") == Some(&Value::I64(order_id)))
961                .filter_map(|row| row.get("id").cloned())
962                .collect::<Vec<_>>();
963            assert_eq!(
964                ids,
965                vec![
966                    Value::I64(order_id * 100 + 5),
967                    Value::I64(order_id * 100 + 4),
968                    Value::I64(order_id * 100 + 3),
969                ]
970            );
971        }
972    }
973
974    #[test]
975    fn sqlite_boolean_new_schema_roundtrips_as_bool() {
976        let executor =
977            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
978        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
979        let ddl = SqliteDialect.compile_create_table(&entity).unwrap();
980        assert!(ddl.contains("enabled BOOLEAN NOT NULL"), "{ddl}");
981        assert!(ddl.contains("optional_enabled BOOLEAN"), "{ddl}");
982        assert!(!ddl.contains("enabled INTEGER"), "{ddl}");
983
984        executor.ensure_schema(&SqliteDialect, &[&entity]).unwrap();
985        for (id, enabled, optional_enabled) in [(1_u64, false, true), (2_u64, true, false)] {
986            let insert = SqliteDialect
987                .compile_insert(
988                    &entity,
989                    &InsertCommand::new("FeatureFlag")
990                        .value("id", id)
991                        .value("version", 1_i64)
992                        .value("enabled", enabled)
993                        .value("optional_enabled", optional_enabled),
994                )
995                .unwrap();
996            assert_eq!(executor.execute(&insert).unwrap(), 1);
997        }
998
999        let select = SqliteDialect
1000            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
1001            .unwrap();
1002        let rows = executor.fetch_all(&select).unwrap();
1003        assert_eq!(rows[0].get("enabled"), Some(&Value::Bool(false)));
1004        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::Bool(true)));
1005        assert_eq!(rows[1].get("enabled"), Some(&Value::Bool(true)));
1006        assert_eq!(rows[1].get("optional_enabled"), Some(&Value::Bool(false)));
1007
1008        let first = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[0].clone()).unwrap();
1009        let second = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[1].clone()).unwrap();
1010        assert!(!first.enabled);
1011        assert_eq!(first.optional_enabled, Some(true));
1012        assert!(second.enabled);
1013        assert_eq!(second.optional_enabled, Some(false));
1014    }
1015
1016    #[test]
1017    fn sqlite_boolean_legacy_integer_schema_maps_only_binary_values() {
1018        let executor =
1019            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1020        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
1021        executor
1022            .execute(&CompiledQuery {
1023                sql: "CREATE TABLE feature_flags (id INTEGER PRIMARY KEY, version INTEGER NOT NULL, enabled INTEGER NOT NULL, optional_enabled INTEGER)"
1024                    .to_owned(),
1025                params: Vec::new(),
1026                comment: None,
1027            })
1028            .unwrap();
1029
1030        let insert = SqliteDialect
1031            .compile_insert(
1032                &entity,
1033                &InsertCommand::new("FeatureFlag")
1034                    .value("id", 1_u64)
1035                    .value("version", 1_i64)
1036                    .value("enabled", true)
1037                    .value("optional_enabled", false),
1038            )
1039            .unwrap();
1040        executor.execute(&insert).unwrap();
1041        executor
1042            .execute(&CompiledQuery {
1043                sql: "INSERT INTO feature_flags (id, version, enabled, optional_enabled) VALUES (?, ?, ?, ?)"
1044                    .to_owned(),
1045                params: vec![
1046                    Value::U64(2),
1047                    Value::I64(1),
1048                    Value::I64(2),
1049                    Value::Null,
1050                ],
1051                comment: None,
1052            })
1053            .unwrap();
1054        let select = SqliteDialect
1055            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
1056            .unwrap();
1057        let rows = executor.fetch_all(&select).unwrap();
1058        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
1059        assert_eq!(rows[0].get("enabled"), Some(&Value::I64(1)));
1060        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::I64(0)));
1061
1062        let decoded = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[0].clone()).unwrap();
1063        assert!(decoded.enabled);
1064        assert_eq!(decoded.optional_enabled, Some(false));
1065        assert_eq!(rows[1].get("enabled"), Some(&Value::I64(2)));
1066        let error =
1067            <FeatureFlagRow as teaql_core::Entity>::from_record(rows[1].clone()).unwrap_err();
1068        assert!(error.message.contains("invalid field enabled"));
1069
1070        for (value, expected) in [
1071            (Value::I64(0), false),
1072            (Value::I64(1), true),
1073            (Value::U64(0), false),
1074            (Value::U64(1), true),
1075        ] {
1076            let decoded = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
1077                value,
1078                Value::Null,
1079            ))
1080            .unwrap();
1081            assert_eq!(decoded.enabled, expected);
1082            assert_eq!(decoded.optional_enabled, None);
1083        }
1084
1085        for invalid in [Value::I64(-1), Value::I64(2), Value::U64(2)] {
1086            let error = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
1087                invalid,
1088                Value::Null,
1089            ))
1090            .unwrap_err();
1091            assert!(error.message.contains("invalid field enabled"));
1092        }
1093        let error = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
1094            Value::Bool(true),
1095            Value::U64(2),
1096        ))
1097        .unwrap_err();
1098        assert!(error.message.contains("invalid field optional_enabled"));
1099    }
1100
1101    #[test]
1102    fn sqlite_executor_parses_json_only_for_json_columns() {
1103        let executor =
1104            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1105
1106        executor
1107            .execute(&CompiledQuery {
1108                sql: "CREATE TABLE payloads (text_payload TEXT, json_payload JSON)".to_owned(),
1109                params: Vec::new(),
1110                comment: None,
1111            })
1112            .unwrap();
1113        executor
1114            .execute(&CompiledQuery {
1115                sql: "INSERT INTO payloads (text_payload, json_payload) VALUES (?, ?)".to_owned(),
1116                params: vec![
1117                    Value::Text("{\"active\":true}".to_owned()),
1118                    Value::Json(serde_json::json!({"active": true})),
1119                ],
1120                comment: None,
1121            })
1122            .unwrap();
1123
1124        let rows = executor
1125            .fetch_all(&CompiledQuery {
1126                sql: "SELECT text_payload, json_payload FROM payloads".to_owned(),
1127                params: Vec::new(),
1128                comment: None,
1129            })
1130            .unwrap();
1131
1132        assert_eq!(
1133            rows[0].get("text_payload"),
1134            Some(&Value::Text("{\"active\":true}".to_owned()))
1135        );
1136        assert_eq!(
1137            rows[0].get("json_payload"),
1138            Some(&Value::Json(serde_json::json!({"active": true})))
1139        );
1140    }
1141
1142    #[test]
1143    fn sqlite_id_space_generator_increments_ids() {
1144        let executor =
1145            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1146        let generator = SqliteIdSpaceGenerator::from_executor(executor);
1147        assert_eq!(generator.next_id("Order").unwrap(), 1);
1148        assert_eq!(generator.next_id("Order").unwrap(), 2);
1149    }
1150
1151    #[test]
1152    fn sqlite_fetch_stream_returns_chunked_rows() {
1153        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1154            Connection::open_in_memory().unwrap(),
1155        )));
1156        let entity = entity();
1157
1158        // Create table and insert 25 rows
1159        executor
1160            .execute(&CompiledQuery {
1161                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1162                    .to_owned(),
1163                params: Vec::new(),
1164                comment: None,
1165            })
1166            .unwrap();
1167
1168        for i in 1..=25 {
1169            let insert = SqliteDialect
1170                .compile_insert(
1171                    &entity,
1172                    &InsertCommand::new("Order")
1173                        .value("id", i as u64)
1174                        .value("version", 1_i64)
1175                        .value("name", format!("order-{i}")),
1176                )
1177                .unwrap();
1178            executor.execute(&insert).unwrap();
1179        }
1180
1181        // Stream with chunk_size = 10
1182        let query = SelectQuery::new("Order")
1183            .filter(Expr::gt("version", 0_i64))
1184            .order_asc("id")
1185            .stream(10);
1186
1187        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1188
1189        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1190
1191        // 25 rows / 10 per chunk = 3 chunks
1192        assert_eq!(chunks.len(), 3);
1193        assert_eq!(chunks[0].rows.len(), 10);
1194        assert_eq!(chunks[0].chunk_index, 0);
1195        assert!(!chunks[0].is_last);
1196
1197        assert_eq!(chunks[1].rows.len(), 10);
1198        assert_eq!(chunks[1].chunk_index, 1);
1199        assert!(!chunks[1].is_last);
1200
1201        assert_eq!(chunks[2].rows.len(), 5);
1202        assert_eq!(chunks[2].chunk_index, 2);
1203        assert!(chunks[2].is_last);
1204
1205        // Verify first and last row
1206        assert_eq!(
1207            chunks[0].rows[0].get("name"),
1208            Some(&Value::Text("order-1".to_owned()))
1209        );
1210        assert_eq!(
1211            chunks[2].rows[4].get("name"),
1212            Some(&Value::Text("order-25".to_owned()))
1213        );
1214    }
1215
1216    #[test]
1217    fn sqlite_fetch_stream_handles_empty_result() {
1218        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1219            Connection::open_in_memory().unwrap(),
1220        )));
1221
1222        executor
1223            .execute(&CompiledQuery {
1224                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1225                    .to_owned(),
1226                params: Vec::new(),
1227                comment: None,
1228            })
1229            .unwrap();
1230
1231        let entity = entity();
1232        let query = SelectQuery::new("Order")
1233            .filter(Expr::gt("version", 0_i64))
1234            .stream(10);
1235
1236        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1237
1238        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1239
1240        // Empty result = 1 chunk with 0 rows, marked as last
1241        assert_eq!(chunks.len(), 1);
1242        assert_eq!(chunks[0].rows.len(), 0);
1243        assert!(chunks[0].is_last);
1244    }
1245
1246    #[test]
1247    fn sqlite_fetch_stream_exact_chunk_boundary() {
1248        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1249            Connection::open_in_memory().unwrap(),
1250        )));
1251        let entity = entity();
1252
1253        executor
1254            .execute(&CompiledQuery {
1255                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1256                    .to_owned(),
1257                params: Vec::new(),
1258                comment: None,
1259            })
1260            .unwrap();
1261
1262        // Insert exactly 20 rows
1263        for i in 1..=20 {
1264            let insert = SqliteDialect
1265                .compile_insert(
1266                    &entity,
1267                    &InsertCommand::new("Order")
1268                        .value("id", i as u64)
1269                        .value("version", 1_i64)
1270                        .value("name", format!("order-{i}")),
1271                )
1272                .unwrap();
1273            executor.execute(&insert).unwrap();
1274        }
1275
1276        let query = SelectQuery::new("Order")
1277            .filter(Expr::gt("version", 0_i64))
1278            .order_asc("id")
1279            .stream(10);
1280
1281        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1282
1283        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1284
1285        // 20 rows / 10 per chunk = 2 full chunks + 1 empty final chunk
1286        assert_eq!(chunks.len(), 3);
1287        assert_eq!(chunks[0].rows.len(), 10);
1288        assert!(!chunks[0].is_last);
1289        assert_eq!(chunks[1].rows.len(), 10);
1290        assert!(!chunks[1].is_last);
1291        assert_eq!(chunks[2].rows.len(), 0);
1292        assert!(chunks[2].is_last);
1293    }
1294
1295    #[test]
1296    fn test_parse_sqlite_timestamp() {
1297        let ts1 = parse_sqlite_timestamp("2023-01-01 12:30:45").unwrap();
1298        assert!(matches!(ts1, Value::Timestamp(_)));
1299
1300        let ts2 = parse_sqlite_timestamp("2023-01-01").unwrap();
1301        assert!(matches!(ts2, Value::Timestamp(_)));
1302
1303        let ts3 = parse_sqlite_timestamp("2023-01-01T12:30:45Z").unwrap();
1304        assert!(matches!(ts3, Value::Timestamp(_)));
1305
1306        assert!(parse_sqlite_timestamp("invalid").is_err());
1307    }
1308}
teaql_provider_sqlite/lib.rs

teaql_provider_sqlite/
lib.rs
