Skip to main content

corium_sql/
mutation.rs

1//! Translation of SQL DML into ordinary Corium transaction forms.
2
3use std::collections::{BTreeMap, BTreeSet};
4
5use corium_core::{Attribute, Cardinality, EntityId, Keyword, TotalF64, Value, ValueType};
6use corium_db::{Db, DbView};
7use corium_query::edn::Edn;
8use sqlparser::ast::{
9    AssignmentTarget, Delete, FromTable, Insert, ObjectName, Query, SelectItem, Statement,
10    TableFactor, Update,
11};
12use sqlparser::dialect::PostgreSqlDialect;
13use sqlparser::parser::Parser;
14
15use crate::{SqlColumn, SqlError, SqlRow, SqlSession, SqlValue};
16
17/// Kind of committed SQL mutation.
18#[derive(Clone, Copy, Debug, Eq, PartialEq)]
19pub enum MutationKind {
20    /// `INSERT`.
21    Insert,
22    /// `UPDATE`.
23    Update,
24    /// `DELETE`.
25    Delete,
26}
27
28impl MutationKind {
29    /// `PostgreSQL` command tag name.
30    #[must_use]
31    pub const fn command(self) -> &'static str {
32        match self {
33            Self::Insert => "INSERT",
34            Self::Update => "UPDATE",
35            Self::Delete => "DELETE",
36        }
37    }
38}
39
40/// Buffered rows produced by a mutation's `RETURNING` clause.
41#[derive(Clone, Debug, Default)]
42pub struct SqlMutationResult {
43    /// Result columns, empty when the statement had no `RETURNING`.
44    pub columns: Vec<SqlColumn>,
45    /// Returned rows.
46    pub rows: Vec<SqlRow>,
47}
48
49#[derive(Clone, Debug)]
50enum EntitySelector {
51    Id(EntityId),
52    Temp(String),
53}
54
55/// One autocommit mutation planned against an immutable basis.
56pub struct SqlMutation {
57    kind: MutationKind,
58    expected_basis_t: u64,
59    forms: Vec<Edn>,
60    affected: usize,
61    table: String,
62    entities: Vec<EntitySelector>,
63    returning: Option<String>,
64    returning_before: Option<SqlMutationResult>,
65    params: Vec<SqlValue>,
66}
67
68impl SqlMutation {
69    /// Mutation kind.
70    #[must_use]
71    pub const fn kind(&self) -> MutationKind {
72        self.kind
73    }
74
75    /// Basis this read/modify/write plan requires.
76    #[must_use]
77    pub const fn expected_basis_t(&self) -> u64 {
78        self.expected_basis_t
79    }
80
81    /// Number of rows selected or inserted by the statement.
82    #[must_use]
83    pub const fn affected(&self) -> usize {
84        self.affected
85    }
86
87    /// Transaction forms to submit through Corium's normal authenticated and
88    /// authorized transactor path.
89    #[must_use]
90    pub fn forms(&self) -> &[Edn] {
91        &self.forms
92    }
93
94    /// Whether committing can be skipped because the statement produced no
95    /// changes. This is possible for an `UPDATE` or `DELETE` matching no rows.
96    #[must_use]
97    pub fn is_empty(&self) -> bool {
98        self.forms.is_empty()
99    }
100
101    /// Describes the columns produced by `RETURNING` without committing.
102    ///
103    /// # Errors
104    /// Returns [`SqlError`] if the returning projection cannot be planned.
105    pub async fn returning_columns(&self, db: &Db) -> Result<Vec<SqlColumn>, SqlError> {
106        let Some(returning) = &self.returning else {
107            return Ok(Vec::new());
108        };
109        let query = SqlSession::new(db)?
110            .query_params(
111                &format!("SELECT {returning} FROM {} WHERE FALSE", self.table),
112                &self.params,
113            )
114            .await?;
115        Ok(query.columns().to_vec())
116    }
117
118    /// Produces `RETURNING` rows from the committed database value.
119    ///
120    /// `DELETE RETURNING` is buffered from the pre-commit snapshot. Inserts
121    /// and updates are read from `db_after`, resolving generated entity ids
122    /// through `tempids`.
123    ///
124    /// # Errors
125    /// Returns [`SqlError`] if a generated id is absent or a returning query
126    /// cannot be executed.
127    pub async fn finish(
128        &self,
129        db_after: &Db,
130        tempids: &BTreeMap<String, EntityId>,
131    ) -> Result<SqlMutationResult, SqlError> {
132        if let Some(result) = &self.returning_before {
133            return Ok(result.clone());
134        }
135        let Some(returning) = &self.returning else {
136            return Ok(SqlMutationResult::default());
137        };
138        let session = SqlSession::new(db_after)?;
139        let mut result = SqlMutationResult::default();
140        let entities = self
141            .entities
142            .iter()
143            .map(|selector| {
144                Ok(match selector {
145                    EntitySelector::Id(entity) => *entity,
146                    EntitySelector::Temp(temp) => *tempids.get(temp).ok_or_else(|| {
147                        SqlError::Mutation(format!(
148                            "transactor did not resolve SQL tempid {temp:?}"
149                        ))
150                    })?,
151                })
152            })
153            .collect::<Result<Vec<_>, SqlError>>()?;
154        for chunk in entities.chunks(1_024) {
155            let ids = chunk
156                .iter()
157                .map(|entity| entity.raw().to_string())
158                .collect::<Vec<_>>()
159                .join(", ");
160            let sql = format!("SELECT {returning} FROM {} WHERE e IN ({ids})", self.table);
161            let query = session.query_params(&sql, &self.params).await?;
162            if result.columns.is_empty() {
163                result.columns = query.columns().to_vec();
164            }
165            result.rows.extend(query.collect().await?);
166        }
167        if result.columns.is_empty() {
168            result.columns = self.returning_columns(db_after).await?;
169        }
170        Ok(result)
171    }
172}
173
174#[derive(Clone)]
175struct Projected {
176    id: EntityId,
177    ident: Keyword,
178    attribute: Attribute,
179}
180
181struct Target {
182    sql_name: String,
183    attributes: BTreeMap<String, Projected>,
184}
185
186pub(crate) async fn plan(
187    db: &Db,
188    sql: &str,
189    params: &[SqlValue],
190) -> Result<Option<SqlMutation>, SqlError> {
191    if db.view() != DbView::Current {
192        return Err(SqlError::Mutation(
193            "writes require a current database view".into(),
194        ));
195    }
196    let mut statements = Parser::parse_sql(&PostgreSqlDialect {}, sql)?;
197    if statements.len() != 1 {
198        return Err(SqlError::Mutation(
199            "one mutation statement is required".into(),
200        ));
201    }
202    match statements.pop().expect("one statement") {
203        Statement::Insert(insert) => plan_insert(db, insert, params).await.map(Some),
204        Statement::Update(update) => plan_update(db, update, params).await.map(Some),
205        Statement::Delete(delete) => plan_delete(db, delete, params).await.map(Some),
206        _ => Ok(None),
207    }
208}
209
210pub(crate) async fn describe(
211    db: &Db,
212    sql: &str,
213    params: &[SqlValue],
214) -> Result<Option<Vec<SqlColumn>>, SqlError> {
215    if db.view() != DbView::Current {
216        return Err(SqlError::Mutation(
217            "writes require a current database view".into(),
218        ));
219    }
220    let statements = Parser::parse_sql(&PostgreSqlDialect {}, sql)?;
221    let [statement] = statements.as_slice() else {
222        return Err(SqlError::Mutation(
223            "one mutation statement is required".into(),
224        ));
225    };
226    let (table_name, returning) = match statement {
227        Statement::Insert(insert) => {
228            let sqlparser::ast::TableObject::TableName(table_name) = &insert.table else {
229                return Err(unsupported("INSERT target must be a table"));
230            };
231            (table_name, insert.returning.as_deref())
232        }
233        Statement::Update(update) => (
234            table_factor_name(&update.table.relation)?,
235            update.returning.as_deref(),
236        ),
237        Statement::Delete(delete) => {
238            let tables = match &delete.from {
239                FromTable::WithFromKeyword(tables) | FromTable::WithoutKeyword(tables) => tables,
240            };
241            let [table] = tables.as_slice() else {
242                return Err(unsupported("DELETE requires exactly one table"));
243            };
244            (
245                table_factor_name(&table.relation)?,
246                delete.returning.as_deref(),
247            )
248        }
249        _ => return Ok(None),
250    };
251    let target = target(db, table_name)?;
252    let Some(returning) = returning_sql(returning) else {
253        return Ok(Some(Vec::new()));
254    };
255    let query = SqlSession::new(db)?
256        .query_params(
257            &format!("SELECT {returning} FROM {} WHERE FALSE", target.sql_name),
258            params,
259        )
260        .await?;
261    Ok(Some(query.columns().to_vec()))
262}
263
264async fn plan_insert(
265    db: &Db,
266    insert: Insert,
267    params: &[SqlValue],
268) -> Result<SqlMutation, SqlError> {
269    let sqlparser::ast::TableObject::TableName(table_name) = &insert.table else {
270        return Err(unsupported("INSERT target must be a table"));
271    };
272    let target = target(db, table_name)?;
273    if insert.columns.is_empty() {
274        return Err(unsupported("INSERT requires an explicit column list"));
275    }
276    if insert.on.is_some()
277        || !insert.assignments.is_empty()
278        || insert.output.is_some()
279        || insert.replace_into
280    {
281        return Err(unsupported(
282            "ON CONFLICT, INSERT SET, OUTPUT, and REPLACE are not supported yet",
283        ));
284    }
285    let source = insert
286        .source
287        .as_deref()
288        .ok_or_else(|| unsupported("INSERT requires VALUES or a query source"))?;
289    let values = evaluate_query(db, source, params).await?;
290    if values.iter().any(|row| row.len() != insert.columns.len()) {
291        return Err(SqlError::Mutation(
292            "INSERT source width does not match its column list".into(),
293        ));
294    }
295
296    let columns = insert
297        .columns
298        .iter()
299        .map(column_name)
300        .collect::<Result<Vec<_>, _>>()?;
301    reject_duplicate_columns(&columns, "INSERT")?;
302    let mut forms = Vec::with_capacity(values.len());
303    let mut entities = Vec::with_capacity(values.len());
304    for (row_index, row) in values.into_iter().enumerate() {
305        let temp = format!("__corium_sql_{row_index}");
306        let mut entity = EntitySelector::Temp(temp.clone());
307        let mut pairs = vec![(Edn::keyword("db/id"), Edn::Str(temp))];
308        for (column, value) in columns.iter().zip(row) {
309            if column == "e" {
310                let id = entity_id(&value)?;
311                if target
312                    .attributes
313                    .values()
314                    .any(|attr| !db.values(id, attr.id).is_empty())
315                {
316                    return Err(SqlError::Mutation(format!(
317                        "entity {} already appears in {}",
318                        id.raw(),
319                        target.sql_name
320                    )));
321                }
322                entity = EntitySelector::Id(id);
323                pairs[0].1 = eid(id)?;
324                continue;
325            }
326            let projected = target
327                .attributes
328                .get(column)
329                .ok_or_else(|| unknown_column(column, &target.sql_name))?;
330            let desired = desired_values(db, projected, &value)?;
331            if desired.is_empty() {
332                continue;
333            }
334            let form = if projected.attribute.cardinality == Cardinality::Many {
335                Edn::Vector(
336                    desired
337                        .iter()
338                        .map(|value| value_to_edn(db, value))
339                        .collect::<Result<_, _>>()?,
340                )
341            } else {
342                value_to_edn(db, &desired[0])?
343            };
344            pairs.push((Edn::Keyword(projected.ident.clone()), form));
345        }
346        if pairs.len() == 1 {
347            return Err(SqlError::Mutation(
348                "INSERT row must assert at least one non-NULL attribute".into(),
349            ));
350        }
351        forms.push(Edn::Map(pairs));
352        entities.push(entity);
353    }
354    Ok(SqlMutation {
355        kind: MutationKind::Insert,
356        expected_basis_t: db.basis_t(),
357        affected: forms.len(),
358        forms,
359        table: target.sql_name,
360        entities,
361        returning: returning_sql(insert.returning.as_deref()),
362        returning_before: None,
363        params: params.to_vec(),
364    })
365}
366
367async fn plan_update(
368    db: &Db,
369    update: Update,
370    params: &[SqlValue],
371) -> Result<SqlMutation, SqlError> {
372    if !update.table.joins.is_empty()
373        || update.from.is_some()
374        || update.output.is_some()
375        || update.limit.is_some()
376        || !update.order_by.is_empty()
377    {
378        return Err(unsupported(
379            "joined/from/ordered/limited UPDATE is not supported yet",
380        ));
381    }
382    let table_name = table_factor_name(&update.table.relation)?;
383    let target = target(db, table_name)?;
384    let mut assignments = Vec::with_capacity(update.assignments.len());
385    let mut assigned = BTreeSet::new();
386    for assignment in &update.assignments {
387        let AssignmentTarget::ColumnName(name) = &assignment.target else {
388            return Err(unsupported("tuple assignment is not supported"));
389        };
390        let column = column_name(name)?;
391        if !assigned.insert(column.clone()) {
392            return Err(SqlError::Mutation(format!(
393                "column {column:?} is assigned more than once"
394            )));
395        }
396        if column == "e" {
397            return Err(SqlError::Mutation("entity column e is immutable".into()));
398        }
399        let projected = target
400            .attributes
401            .get(&column)
402            .ok_or_else(|| unknown_column(&column, &target.sql_name))?
403            .clone();
404        assignments.push((projected, assignment.value.to_string()));
405    }
406    let selection = update
407        .selection
408        .as_ref()
409        .map(|expr| format!(" WHERE {expr}"))
410        .unwrap_or_default();
411    let projections = assignments
412        .iter()
413        .enumerate()
414        .map(|(index, (_, expression))| format!("({expression}) AS \"__set_{index}\""))
415        .collect::<Vec<_>>()
416        .join(", ");
417    let sql = if projections.is_empty() {
418        format!("SELECT e FROM {}{selection}", target.sql_name)
419    } else {
420        format!(
421            "SELECT e, {projections} FROM {}{selection}",
422            target.sql_name
423        )
424    };
425    let rows = evaluate_sql(db, &sql, params).await?;
426    let mut forms = Vec::new();
427    let mut entities = Vec::with_capacity(rows.len());
428    for row in &rows {
429        let entity = entity_id(&row[0])?;
430        entities.push(EntitySelector::Id(entity));
431        for ((projected, _), value) in assignments.iter().zip(&row[1..]) {
432            let desired = desired_values(db, projected, value)?;
433            let current = db.values(entity, projected.id);
434            for old in current.iter().filter(|old| !desired.contains(old)) {
435                forms.push(retract(entity, projected, value_to_edn(db, old)?)?);
436            }
437            for new in desired.iter().filter(|new| !current.contains(new)) {
438                forms.push(add(entity, projected, value_to_edn(db, new)?)?);
439            }
440        }
441    }
442    Ok(SqlMutation {
443        kind: MutationKind::Update,
444        expected_basis_t: db.basis_t(),
445        forms,
446        affected: rows.len(),
447        table: target.sql_name,
448        entities,
449        returning: returning_sql(update.returning.as_deref()),
450        returning_before: None,
451        params: params.to_vec(),
452    })
453}
454
455async fn plan_delete(
456    db: &Db,
457    delete: Delete,
458    params: &[SqlValue],
459) -> Result<SqlMutation, SqlError> {
460    if delete.using.is_some()
461        || delete.output.is_some()
462        || delete.limit.is_some()
463        || !delete.order_by.is_empty()
464        || !delete.tables.is_empty()
465    {
466        return Err(unsupported(
467            "multi-table/using/ordered/limited DELETE is not supported yet",
468        ));
469    }
470    let tables = match &delete.from {
471        FromTable::WithFromKeyword(tables) | FromTable::WithoutKeyword(tables) => tables,
472    };
473    let [table] = tables.as_slice() else {
474        return Err(unsupported("DELETE requires exactly one table"));
475    };
476    if !table.joins.is_empty() {
477        return Err(unsupported("joined DELETE is not supported yet"));
478    }
479    let table_name = table_factor_name(&table.relation)?;
480    let target = target(db, table_name)?;
481    let selection = delete
482        .selection
483        .as_ref()
484        .map(|expr| format!(" WHERE {expr}"))
485        .unwrap_or_default();
486    let entity_rows = evaluate_sql(
487        db,
488        &format!("SELECT e FROM {}{selection}", target.sql_name),
489        params,
490    )
491    .await?;
492    let entity_ids = entity_rows
493        .iter()
494        .map(|row| entity_id(&row[0]))
495        .collect::<Result<Vec<_>, _>>()?;
496    let mut forms = Vec::new();
497    for entity in &entity_ids {
498        for projected in target.attributes.values() {
499            for old in db.values(*entity, projected.id) {
500                forms.push(retract(*entity, projected, value_to_edn(db, &old)?)?);
501            }
502        }
503    }
504    let returning = returning_sql(delete.returning.as_deref());
505    let returning_before = match &returning {
506        None => None,
507        Some(items) => Some(
508            evaluate_result(
509                db,
510                &format!("SELECT {items} FROM {}{selection}", target.sql_name),
511                params,
512            )
513            .await?,
514        ),
515    };
516    let entities = entity_ids.iter().copied().map(EntitySelector::Id).collect();
517    Ok(SqlMutation {
518        kind: MutationKind::Delete,
519        expected_basis_t: db.basis_t(),
520        affected: entity_ids.len(),
521        forms,
522        table: target.sql_name,
523        entities,
524        returning,
525        returning_before,
526        params: params.to_vec(),
527    })
528}
529
530fn target(db: &Db, name: &ObjectName) -> Result<Target, SqlError> {
531    let parts = name
532        .0
533        .iter()
534        .map(|part| {
535            part.as_ident()
536                .map(normalized_ident)
537                .ok_or_else(|| unsupported("dynamic table names are not supported"))
538        })
539        .collect::<Result<Vec<_>, _>>()?;
540    let [schema, namespace] = parts.as_slice() else {
541        return Err(SqlError::Mutation(
542            "writable tables must be named corium.<namespace>".into(),
543        ));
544    };
545    if schema != "corium" {
546        return Err(SqlError::Mutation(
547            "only corium namespace projections are writable".into(),
548        ));
549    }
550    let actual_namespace = (namespace != "_global").then_some(namespace.as_str());
551    let mut attributes = BTreeMap::new();
552    for (id, attribute) in db.schema().iter() {
553        let Some(ident) = db.idents().ident(*id) else {
554            continue;
555        };
556        if ident.namespace.as_deref() == actual_namespace {
557            attributes.insert(
558                ident.name.clone(),
559                Projected {
560                    id: *id,
561                    ident: ident.clone(),
562                    attribute: attribute.clone(),
563                },
564            );
565        }
566    }
567    if attributes.is_empty() {
568        return Err(SqlError::Mutation(format!(
569            "writable projection {name} does not exist"
570        )));
571    }
572    Ok(Target {
573        sql_name: name.to_string(),
574        attributes,
575    })
576}
577
578fn table_factor_name(factor: &TableFactor) -> Result<&ObjectName, SqlError> {
579    match factor {
580        TableFactor::Table { name, args, .. } if args.is_none() => Ok(name),
581        _ => Err(unsupported("mutation target must be a plain table")),
582    }
583}
584
585fn column_name(name: &ObjectName) -> Result<String, SqlError> {
586    let [part] = name.0.as_slice() else {
587        return Err(unsupported("qualified mutation columns are not supported"));
588    };
589    part.as_ident()
590        .map(normalized_ident)
591        .ok_or_else(|| unsupported("dynamic column names are not supported"))
592}
593
594fn normalized_ident(ident: &sqlparser::ast::Ident) -> String {
595    if ident.quote_style.is_none() {
596        ident.value.to_ascii_lowercase()
597    } else {
598        ident.value.clone()
599    }
600}
601
602fn reject_duplicate_columns(columns: &[String], statement: &str) -> Result<(), SqlError> {
603    let mut seen = BTreeSet::new();
604    for column in columns {
605        if !seen.insert(column) {
606            return Err(SqlError::Mutation(format!(
607                "{statement} column {column:?} is specified more than once"
608            )));
609        }
610    }
611    Ok(())
612}
613
614fn desired_values(
615    db: &Db,
616    projected: &Projected,
617    value: &SqlValue,
618) -> Result<Vec<Value>, SqlError> {
619    match projected.attribute.cardinality {
620        Cardinality::One => match value {
621            SqlValue::Null => Ok(Vec::new()),
622            SqlValue::List(_) => Err(type_error(projected, "a scalar")),
623            value => Ok(vec![sql_value(db, projected, value)?]),
624        },
625        Cardinality::Many => match value {
626            SqlValue::List(values) => {
627                let mut result = Vec::with_capacity(values.len());
628                for value in values {
629                    if matches!(value, SqlValue::Null) {
630                        return Err(type_error(projected, "a list without NULL elements"));
631                    }
632                    let value = sql_value(db, projected, value)?;
633                    if !result.contains(&value) {
634                        result.push(value);
635                    }
636                }
637                result.sort();
638                Ok(result)
639            }
640            _ => Err(type_error(projected, "an ARRAY")),
641        },
642    }
643}
644
645fn sql_value(db: &Db, projected: &Projected, value: &SqlValue) -> Result<Value, SqlError> {
646    if let SqlValue::Unspecified(value) = value {
647        return sql_unspecified(db, projected, value);
648    }
649    // DataFusion resolves an unknown placeholder used as a standalone
650    // assignment expression to UTF-8. Apply PostgreSQL-style assignment
651    // coercion for non-text targets after evaluating that expression.
652    if let SqlValue::Text(value) = value
653        && !matches!(
654            projected.attribute.value_type,
655            ValueType::Uuid | ValueType::Keyword | ValueType::Str
656        )
657    {
658        return sql_unspecified(db, projected, value);
659    }
660    let out = match (projected.attribute.value_type, value) {
661        (ValueType::Bool, SqlValue::Boolean(value)) => Value::Bool(*value),
662        (ValueType::Long, SqlValue::Integer(value)) => Value::Long(*value),
663        (ValueType::Long, SqlValue::Unsigned(value)) => {
664            Value::Long(i64::try_from(*value).map_err(|_| type_error(projected, "a BIGINT"))?)
665        }
666        (ValueType::Double, SqlValue::Float(value)) => Value::Double(TotalF64(*value)),
667        (ValueType::Double, SqlValue::Integer(value)) => {
668            #[allow(clippy::cast_precision_loss)]
669            let value = *value as f64;
670            Value::Double(TotalF64(value))
671        }
672        (ValueType::Double, SqlValue::Unsigned(value)) => {
673            #[allow(clippy::cast_precision_loss)]
674            let value = *value as f64;
675            Value::Double(TotalF64(value))
676        }
677        (ValueType::Instant, SqlValue::TimestampMillis(value) | SqlValue::Integer(value)) => {
678            Value::Instant(*value)
679        }
680        (ValueType::Instant, SqlValue::Unsigned(value)) => {
681            Value::Instant(i64::try_from(*value).map_err(|_| type_error(projected, "a TIMESTAMP"))?)
682        }
683        (ValueType::Uuid, SqlValue::Text(value)) => {
684            Value::Uuid(parse_uuid(value).ok_or_else(|| type_error(projected, "a UUID string"))?)
685        }
686        (ValueType::Keyword, SqlValue::Text(value)) => {
687            let keyword = Keyword::parse(value.strip_prefix(':').unwrap_or(value));
688            let id = db.interner().get(&keyword).ok_or_else(|| {
689                SqlError::Mutation(format!(
690                    "keyword {keyword} is not interned yet; SQL keyword insertion is not supported"
691                ))
692            })?;
693            Value::Keyword(id)
694        }
695        (ValueType::Str, SqlValue::Text(value)) => Value::Str(value.as_str().into()),
696        (ValueType::Bytes, SqlValue::Bytes(value)) => Value::Bytes(value.clone().into()),
697        (ValueType::Ref, SqlValue::Unsigned(value)) => Value::Ref(EntityId::from_raw(*value)),
698        (ValueType::Ref, SqlValue::Integer(value)) if *value >= 0 => Value::Ref(
699            EntityId::from_raw(u64::try_from(*value).expect("nonnegative")),
700        ),
701        _ => {
702            return Err(type_error(
703                projected,
704                value_type_description(projected.attribute.value_type),
705            ));
706        }
707    };
708    Ok(out)
709}
710
711fn sql_unspecified(db: &Db, projected: &Projected, value: &str) -> Result<Value, SqlError> {
712    let parsed = match projected.attribute.value_type {
713        ValueType::Bool => match value.to_ascii_lowercase().as_str() {
714            "t" | "true" | "1" => SqlValue::Boolean(true),
715            "f" | "false" | "0" => SqlValue::Boolean(false),
716            _ => return Err(type_error(projected, "a BOOLEAN")),
717        },
718        ValueType::Long => SqlValue::Integer(
719            value
720                .parse()
721                .map_err(|_| type_error(projected, "a BIGINT"))?,
722        ),
723        ValueType::Double => SqlValue::Float(
724            value
725                .parse()
726                .map_err(|_| type_error(projected, "a DOUBLE"))?,
727        ),
728        ValueType::Instant => SqlValue::TimestampMillis(
729            value
730                .parse()
731                .map_err(|_| type_error(projected, "epoch milliseconds"))?,
732        ),
733        ValueType::Uuid | ValueType::Keyword | ValueType::Str => SqlValue::Text(value.into()),
734        ValueType::Bytes => {
735            let hex = value
736                .strip_prefix("\\x")
737                .ok_or_else(|| type_error(projected, "hex BYTEA text"))?;
738            if !hex.len().is_multiple_of(2) || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) {
739                return Err(type_error(projected, "hex BYTEA text"));
740            }
741            let bytes = hex
742                .as_bytes()
743                .chunks_exact(2)
744                .map(|pair| {
745                    u8::from_str_radix(std::str::from_utf8(pair).expect("ASCII hex checked"), 16)
746                        .expect("hex digits checked")
747                })
748                .collect();
749            SqlValue::Bytes(bytes)
750        }
751        ValueType::Ref => SqlValue::Unsigned(
752            value
753                .parse()
754                .map_err(|_| type_error(projected, "an entity id"))?,
755        ),
756    };
757    sql_value(db, projected, &parsed)
758}
759
760fn parse_uuid(value: &str) -> Option<u128> {
761    let bytes = value.as_bytes();
762    let valid_shape = (bytes.len() == 32 && bytes.iter().all(u8::is_ascii_hexdigit))
763        || (bytes.len() == 36
764            && bytes.iter().enumerate().all(|(index, byte)| match index {
765                8 | 13 | 18 | 23 => *byte == b'-',
766                _ => byte.is_ascii_hexdigit(),
767            }));
768    if !valid_shape {
769        return None;
770    }
771    let compact = value.replace('-', "");
772    (compact.len() == 32 && compact.bytes().all(|byte| byte.is_ascii_hexdigit()))
773        .then(|| u128::from_str_radix(&compact, 16).ok())
774        .flatten()
775}
776
777fn value_to_edn(db: &Db, value: &Value) -> Result<Edn, SqlError> {
778    Ok(match value {
779        Value::Bool(value) => Edn::Bool(*value),
780        Value::Long(value) => Edn::Long(*value),
781        Value::Double(value) => Edn::Double(*value),
782        Value::Instant(value) => Edn::Tagged("inst".into(), Box::new(Edn::Long(*value))),
783        Value::Uuid(value) => {
784            Edn::Tagged("uuid".into(), Box::new(Edn::Str(format!("{value:032x}"))))
785        }
786        Value::Keyword(id) => {
787            let keyword = db.interner().resolve(*id).ok_or_else(|| {
788                SqlError::Mutation("new keyword value lost its SQL spelling".into())
789            })?;
790            Edn::Keyword(keyword.clone())
791        }
792        Value::Str(value) => Edn::Str(value.to_string()),
793        Value::Bytes(value) => {
794            use std::fmt::Write as _;
795
796            let hex = value
797                .iter()
798                .fold(String::with_capacity(value.len() * 2), |mut hex, byte| {
799                    let _ = write!(hex, "{byte:02x}");
800                    hex
801                });
802            Edn::Tagged("bytes".into(), Box::new(Edn::Str(hex)))
803        }
804        Value::Ref(value) => return eid(*value),
805    })
806}
807
808fn add(entity: EntityId, projected: &Projected, value: Edn) -> Result<Edn, SqlError> {
809    Ok(Edn::Vector(vec![
810        Edn::keyword("db/add"),
811        eid(entity)?,
812        Edn::Keyword(projected.ident.clone()),
813        value,
814    ]))
815}
816
817fn retract(entity: EntityId, projected: &Projected, value: Edn) -> Result<Edn, SqlError> {
818    Ok(Edn::Vector(vec![
819        Edn::keyword("db/retract"),
820        eid(entity)?,
821        Edn::Keyword(projected.ident.clone()),
822        value,
823    ]))
824}
825
826fn eid(entity: EntityId) -> Result<Edn, SqlError> {
827    let raw = i64::try_from(entity.raw()).map_err(|_| {
828        SqlError::Mutation(format!(
829            "entity id {} exceeds the SQL mutation boundary",
830            entity.raw()
831        ))
832    })?;
833    Ok(Edn::Tagged("eid".into(), Box::new(Edn::Long(raw))))
834}
835
836fn entity_id(value: &SqlValue) -> Result<EntityId, SqlError> {
837    match value {
838        SqlValue::Unsigned(value) => Ok(EntityId::from_raw(*value)),
839        SqlValue::Integer(value) if *value >= 0 => Ok(EntityId::from_raw(
840            u64::try_from(*value).expect("nonnegative"),
841        )),
842        _ => Err(SqlError::Mutation(
843            "entity column e requires a non-negative integer".into(),
844        )),
845    }
846}
847
848async fn evaluate_query(
849    db: &Db,
850    query: &Query,
851    params: &[SqlValue],
852) -> Result<Vec<SqlRow>, SqlError> {
853    evaluate_sql(db, &query.to_string(), params).await
854}
855
856async fn evaluate_sql(db: &Db, sql: &str, params: &[SqlValue]) -> Result<Vec<SqlRow>, SqlError> {
857    SqlSession::new(db)?
858        .query_params(sql, params)
859        .await?
860        .collect()
861        .await
862}
863
864async fn evaluate_result(
865    db: &Db,
866    sql: &str,
867    params: &[SqlValue],
868) -> Result<SqlMutationResult, SqlError> {
869    let query = SqlSession::new(db)?.query_params(sql, params).await?;
870    let columns = query.columns().to_vec();
871    let rows = query.collect().await?;
872    Ok(SqlMutationResult { columns, rows })
873}
874
875fn returning_sql(items: Option<&[SelectItem]>) -> Option<String> {
876    items.map(|items| {
877        items
878            .iter()
879            .map(ToString::to_string)
880            .collect::<Vec<_>>()
881            .join(", ")
882    })
883}
884
885fn unsupported(message: &str) -> SqlError {
886    SqlError::Mutation(message.into())
887}
888
889fn unknown_column(column: &str, table: &str) -> SqlError {
890    SqlError::Mutation(format!("column {column:?} does not exist in {table}"))
891}
892
893fn type_error(projected: &Projected, expected: &str) -> SqlError {
894    SqlError::Mutation(format!("attribute {} expects {expected}", projected.ident))
895}
896
897const fn value_type_description(value_type: ValueType) -> &'static str {
898    match value_type {
899        ValueType::Bool => "a BOOLEAN",
900        ValueType::Long => "a BIGINT",
901        ValueType::Double => "a DOUBLE",
902        ValueType::Instant => "a TIMESTAMP",
903        ValueType::Uuid => "a UUID string",
904        ValueType::Keyword => "keyword text",
905        ValueType::Str => "TEXT",
906        ValueType::Bytes => "BYTEA",
907        ValueType::Ref => "an entity id",
908    }
909}