teaql_provider_postgres/
lib.rs

1#![allow(warnings)]
2use std::collections::BTreeMap;
3use std::future::Future;
4use std::pin::Pin;
5
6use chrono::{DateTime, NaiveDate, Utc};
7use deadpool_postgres::Pool;
8use rust_decimal::Decimal;
9use std::sync::Arc;
10use teaql_core::{
11    BinaryOp, DataType, EntityDescriptor, Expr, InsertCommand, PropertyDescriptor, Record,
12    SelectQuery, UpdateCommand, Value,
13};
14use teaql_runtime::{GraphNode, InternalIdGenerator, RuntimeError, SchemaProvider, UserContext};
15use teaql_sql::{
16    CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, SqlTransport,
17    quote_identifier_if_needed,
18};
19use tokio::sync::Mutex;
20
21pub const DEFAULT_ID_SPACE_TABLE: &str = "teaql_id_space";
22
23#[derive(Debug, Default, Clone, Copy)]
24pub struct PostgresDialect;
25
26impl SqlDialect for PostgresDialect {
27    fn kind(&self) -> DatabaseKind {
28        DatabaseKind::PostgreSql
29    }
30
31    fn quote_ident(&self, ident: &str) -> String {
32        quote_ident(ident)
33    }
34
35    fn placeholder(&self, index: usize) -> String {
36        format!("${index}")
37    }
38
39    fn schema_setup_sqls(&self) -> &'static [&'static str] {
40        &[CREATE_SOUNDEX_FUNCTION]
41    }
42
43    fn schema_type_sql(
44        &self,
45        data_type: DataType,
46        _property: &PropertyDescriptor,
47    ) -> Result<&'static str, SqlCompileError> {
48        match data_type {
49            DataType::Bool => Ok("BOOLEAN"),
50            DataType::I64 | DataType::U64 => Ok("BIGINT"),
51            DataType::F64 => Ok("DOUBLE PRECISION"),
52            DataType::Decimal => Ok("NUMERIC"),
53            DataType::Text => Ok("TEXT"),
54            DataType::Json => Ok("JSONB"),
55            DataType::Date => Ok("DATE"),
56            DataType::Timestamp => Ok("TIMESTAMPTZ"),
57        }
58    }
59
60    fn compile_in(
61        &self,
62        entity: &EntityDescriptor,
63        left: &Expr,
64        op: BinaryOp,
65        right: &Expr,
66        params: &mut Vec<Value>,
67    ) -> Result<String, SqlCompileError> {
68        match op {
69            BinaryOp::InLarge | BinaryOp::NotInLarge => {
70                let Expr::Value(Value::List(values)) = right else {
71                    let lhs = self.compile_expr(entity, left, params)?;
72                    let rhs = self.compile_expr(entity, right, params)?;
73                    let operator = match op {
74                        BinaryOp::InLarge => "= ANY",
75                        BinaryOp::NotInLarge => "<> ALL",
76                        _ => unreachable!(),
77                    };
78                    return Ok(format!("({lhs} {operator} ({rhs}))"));
79                };
80                if values.is_empty() {
81                    return Err(SqlCompileError::EmptyInList);
82                }
83                let lhs = self.compile_expr(entity, left, params)?;
84                params.push(Value::List(values.clone()));
85                let placeholder = self.placeholder(params.len());
86                let operator = match op {
87                    BinaryOp::InLarge => "= ANY",
88                    BinaryOp::NotInLarge => "<> ALL",
89                    _ => unreachable!(),
90                };
91                Ok(format!("({lhs} {operator}({placeholder}))"))
92            }
93            _ => {
94                let lhs = self.compile_expr(entity, left, params)?;
95                let operator = match op {
96                    BinaryOp::In => "IN",
97                    BinaryOp::NotIn => "NOT IN",
98                    _ => unreachable!(),
99                };
100                match right {
101                    Expr::Value(Value::List(values)) => {
102                        if values.is_empty() {
103                            return Err(SqlCompileError::EmptyInList);
104                        }
105                        let mut placeholders = Vec::with_capacity(values.len());
106                        for value in values {
107                            params.push(value.clone());
108                            placeholders.push(self.placeholder(params.len()));
109                        }
110                        Ok(format!("({lhs} {operator} ({}))", placeholders.join(", ")))
111                    }
112                    _ => {
113                        let rhs = self.compile_expr(entity, right, params)?;
114                        Ok(format!("({lhs} {operator} ({rhs}))"))
115                    }
116                }
117            }
118        }
119    }
120}
121
122const CREATE_SOUNDEX_FUNCTION: &str = r#"
123CREATE OR REPLACE FUNCTION soundex(input text)
124RETURNS text
125LANGUAGE plpgsql
126IMMUTABLE
127STRICT
128AS $$
129DECLARE
130    normalized text := upper(regexp_replace(input, '[^A-Za-z]', '', 'g'));
131    first_char text;
132    output text;
133    previous_code text;
134    code text;
135    ch text;
136    i integer;
137BEGIN
138    IF normalized = '' THEN
139        RETURN '0000';
140    END IF;
141
142    first_char := substr(normalized, 1, 1);
143    output := first_char;
144    previous_code := CASE
145        WHEN first_char IN ('B', 'F', 'P', 'V') THEN '1'
146        WHEN first_char IN ('C', 'G', 'J', 'K', 'Q', 'S', 'X', 'Z') THEN '2'
147        WHEN first_char IN ('D', 'T') THEN '3'
148        WHEN first_char = 'L' THEN '4'
149        WHEN first_char IN ('M', 'N') THEN '5'
150        WHEN first_char = 'R' THEN '6'
151        ELSE '0'
152    END;
153
154    FOR i IN 2..char_length(normalized) LOOP
155        ch := substr(normalized, i, 1);
156        code := CASE
157            WHEN ch IN ('B', 'F', 'P', 'V') THEN '1'
158            WHEN ch IN ('C', 'G', 'J', 'K', 'Q', 'S', 'X', 'Z') THEN '2'
159            WHEN ch IN ('D', 'T') THEN '3'
160            WHEN ch = 'L' THEN '4'
161            WHEN ch IN ('M', 'N') THEN '5'
162            WHEN ch = 'R' THEN '6'
163            ELSE '0'
164        END;
165
166        IF code <> '0' AND code <> previous_code THEN
167            output := output || code;
168            IF char_length(output) = 4 THEN
169                RETURN output;
170            END IF;
171        END IF;
172        previous_code := code;
173    END LOOP;
174
175    RETURN rpad(output, 4, '0');
176END;
177$$
178"#;
179
180#[derive(Debug)]
181pub enum MutationExecutorError {
182    Driver(tokio_postgres::Error),
183    Pool(String),
184    SqlCompile(SqlCompileError),
185    UnsupportedValue(&'static str),
186    UnsupportedColumnType(String),
187    Bind(String),
188}
189
190impl std::fmt::Display for MutationExecutorError {
191    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
192        match self {
193            Self::Driver(err) => err.fmt(f),
194            Self::Pool(err) => write!(f, "postgres pool error: {err}"),
195            Self::SqlCompile(err) => err.fmt(f),
196            Self::UnsupportedValue(kind) => {
197                write!(f, "unsupported bind value for mutation executor: {kind}")
198            }
199            Self::UnsupportedColumnType(kind) => {
200                write!(f, "unsupported column type for record decoding: {kind}")
201            }
202            Self::Bind(message) => write!(f, "bind error: {message}"),
203        }
204    }
205}
206
207impl std::error::Error for MutationExecutorError {}
208
209impl From<tokio_postgres::Error> for MutationExecutorError {
210    fn from(value: tokio_postgres::Error) -> Self {
211        Self::Driver(value)
212    }
213}
214
215impl From<SqlCompileError> for MutationExecutorError {
216    fn from(value: SqlCompileError) -> Self {
217        Self::SqlCompile(value)
218    }
219}
220
221#[derive(Clone)]
222pub struct PgMutationExecutor {
223    pool: Pool,
224}
225
226impl SqlTransport for PgMutationExecutor {
227    type Error = MutationExecutorError;
228
229    async fn fetch_all_sql(&self, query: &CompiledQuery) -> Result<Vec<Record>, Self::Error> {
230        self.fetch_all(query).await
231    }
232
233    async fn execute_sql(&self, query: &CompiledQuery) -> Result<u64, Self::Error> {
234        self.execute(query).await
235    }
236}
237
238impl teaql_sql::SqlTransaction for PgMutationExecutor {
239    type Error = MutationExecutorError;
240
241    async fn commit_sql(self) -> Result<(), Self::Error> {
242        Err(MutationExecutorError::Bind(
243            "Transactions not supported yet".to_string(),
244        ))
245    }
246
247    async fn rollback_sql(self) -> Result<(), Self::Error> {
248        Err(MutationExecutorError::Bind(
249            "Transactions not supported yet".to_string(),
250        ))
251    }
252}
253
254impl teaql_sql::SqlTransactionTransport for PgMutationExecutor {
255    type Tx<'a>
256        = Self
257    where
258        Self: 'a;
259
260    async fn begin_sql(&self) -> Result<Self::Tx<'_>, Self::Error> {
261        Err(MutationExecutorError::Bind(
262            "Transactions not supported yet".to_string(),
263        ))
264    }
265}
266
267impl PgMutationExecutor {
268    pub fn new(pool: Pool) -> Self {
269        Self { pool }
270    }
271
272    pub fn pool(&self) -> Pool {
273        self.pool.clone()
274    }
275
276    pub async fn ensure_schema(
277        &self,
278        dialect: &PostgresDialect,
279        entities: &[&EntityDescriptor],
280    ) -> Result<(), MutationExecutorError> {
281        let client = self
282            .pool
283            .get()
284            .await
285            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
286        for sql in dialect.schema_setup_sqls() {
287            client.execute(*sql, &[]).await?;
288        }
289        self.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE).await?;
290
291        for entity in entities {
292            if !self.table_exists(&entity.table_name).await? {
293                let sql = dialect.compile_create_table(entity)?;
294                client.execute(&sql, &[]).await?;
295                continue;
296            }
297
298            let existing_columns = self.table_columns(&entity.table_name).await?;
299            for property in &entity.properties {
300                let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
301                if existing_columns.contains(&bare_column) {
302                    continue;
303                }
304                let sql = dialect.compile_add_column(entity, property)?;
305                client.execute(&sql, &[]).await?;
306            }
307
308            for sql in dialect.schema_indexes_sqls(entity)? {
309                client.execute(&sql, &[]).await?;
310            }
311        }
312        Ok(())
313    }
314
315    pub async fn ensure_id_space_table(
316        &self,
317        table_name: &str,
318    ) -> Result<(), MutationExecutorError> {
319        let sql = format!(
320            "CREATE TABLE IF NOT EXISTS {} (type_name VARCHAR(100) PRIMARY KEY, current_level BIGINT NOT NULL)",
321            quote_ident(table_name)
322        );
323        let client = self
324            .pool
325            .get()
326            .await
327            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
328        client.execute(&sql, &[]).await?;
329        Ok(())
330    }
331
332    pub async fn execute(&self, query: &CompiledQuery) -> Result<u64, MutationExecutorError> {
333        let mut args = PgArgs { values: Vec::new() };
334        for value in &query.params {
335            bind_pg(&mut args, value)?;
336        }
337        let client = self
338            .pool
339            .get()
340            .await
341            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
342        let result = client.execute(&query.sql, &args.as_refs()).await?;
343        Ok(result)
344    }
345
346    pub async fn fetch_all(
347        &self,
348        query: &CompiledQuery,
349    ) -> Result<Vec<Record>, MutationExecutorError> {
350        let mut args = PgArgs { values: Vec::new() };
351        for value in &query.params {
352            bind_pg(&mut args, value)?;
353        }
354        let client = self
355            .pool
356            .get()
357            .await
358            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
359        let rows = client.query(&query.sql, &args.as_refs()).await?;
360        rows.iter().map(decode_pg_row).collect()
361    }
362
363    async fn table_exists(&self, table_name: &str) -> Result<bool, MutationExecutorError> {
364        let client = self
365            .pool
366            .get()
367            .await
368            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
369        let row = client
370            .query_one(
371                "SELECT COUNT(1)
372             FROM information_schema.tables
373             WHERE table_schema = current_schema()
374               AND table_name = $1",
375                &[&table_name],
376            )
377            .await?;
378        let exists: i64 = row.try_get(0)?;
379        Ok(exists > 0)
380    }
381
382    async fn table_columns(
383        &self,
384        table_name: &str,
385    ) -> Result<std::collections::BTreeSet<String>, MutationExecutorError> {
386        let client = self
387            .pool
388            .get()
389            .await
390            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
391        let rows = client
392            .query(
393                "SELECT column_name
394             FROM information_schema.columns
395             WHERE table_schema = current_schema()
396               AND table_name = $1",
397                &[&table_name],
398            )
399            .await?;
400        let mut columns = std::collections::BTreeSet::new();
401        for row in rows {
402            let name: String = row.try_get("column_name")?;
403            columns.insert(name.to_lowercase());
404        }
405        Ok(columns)
406    }
407}
408
409async fn ensure_initial_graphs_postgres(
410    executor: &PgMutationExecutor,
411    dialect: &PostgresDialect,
412    ctx: &UserContext,
413) -> Result<(), MutationExecutorError> {
414    for graph in ctx.initial_graphs() {
415        let entity = ctx.entity(&graph.entity).ok_or_else(|| {
416            MutationExecutorError::Bind(format!("missing entity: {}", graph.entity))
417        })?;
418        if initial_graph_exists_postgres(executor, dialect, entity, graph).await? {
419            if let Some(query) = compile_initial_graph_update(dialect, entity, graph)? {
420                executor.execute(&query).await?;
421            }
422            continue;
423        }
424        let query = compile_initial_graph_insert(dialect, entity, graph)?;
425        executor.execute(&query).await?;
426    }
427    Ok(())
428}
429
430async fn initial_graph_exists_postgres(
431    executor: &PgMutationExecutor,
432    dialect: &PostgresDialect,
433    entity: &EntityDescriptor,
434    graph: &GraphNode,
435) -> Result<bool, MutationExecutorError> {
436    let Some(id) = graph.values.get("id") else {
437        return Ok(false);
438    };
439    let query = dialect.compile_select(
440        entity,
441        &SelectQuery::new(&graph.entity)
442            .project("id")
443            .filter(Expr::eq("id", id.clone()))
444            .limit(1),
445    )?;
446    Ok(!executor.fetch_all(&query).await?.is_empty())
447}
448
449fn compile_initial_graph_insert(
450    dialect: &impl SqlDialect,
451    entity: &EntityDescriptor,
452    graph: &GraphNode,
453) -> Result<CompiledQuery, MutationExecutorError> {
454    let mut command = InsertCommand::new(&graph.entity);
455    for (field, value) in &graph.values {
456        command = command.value(field.clone(), value.clone());
457    }
458    dialect.compile_insert(entity, &command).map_err(Into::into)
459}
460
461fn compile_initial_graph_update(
462    dialect: &impl SqlDialect,
463    entity: &EntityDescriptor,
464    graph: &crate::GraphNode,
465) -> Result<Option<CompiledQuery>, MutationExecutorError> {
466    let Some(id) = graph.values.get("id") else {
467        return Ok(None);
468    };
469    let mut command = UpdateCommand::new(&graph.entity, id.clone());
470    for (field, value) in &graph.values {
471        if field == "id" {
472            continue;
473        }
474        command = command.value(field.clone(), value.clone());
475    }
476    match dialect.compile_update(entity, &command) {
477        Ok(query) => Ok(Some(query)),
478        Err(SqlCompileError::EmptyMutation(_)) => Ok(None),
479        Err(err) => Err(err.into()),
480    }
481}
482
483pub trait PostgresSchemaExt {
484    fn ensure_postgres_schema(
485        &self,
486    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + '_>>;
487}
488
489pub async fn ensure_postgres_schema_for(ctx: &UserContext) -> Result<(), MutationExecutorError> {
490    let dialect = ctx.get_resource::<PostgresDialect>().ok_or_else(|| {
491        MutationExecutorError::Bind("missing typed resource: PostgresDialect".to_owned())
492    })?;
493    let executor = ctx.get_resource::<PgMutationExecutor>().ok_or_else(|| {
494        MutationExecutorError::Bind("missing typed resource: PgMutationExecutor".to_owned())
495    })?;
496
497    let entities = ctx.all_entities();
498
499    executor.ensure_schema(dialect, &entities).await?;
500    ensure_initial_graphs_postgres(executor, dialect, ctx).await
501}
502
503impl PostgresSchemaExt for UserContext {
504    fn ensure_postgres_schema(
505        &self,
506    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + '_>> {
507        Box::pin(ensure_postgres_schema_for(self))
508    }
509}
510
511#[derive(Debug, Default, Clone, Copy)]
512pub struct PostgresSchemaProvider;
513
514impl SchemaProvider for PostgresSchemaProvider {
515    fn ensure_schema<'a>(
516        &'a self,
517        ctx: &'a UserContext,
518    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>> {
519        Box::pin(async move {
520            ensure_postgres_schema_for(ctx)
521                .await
522                .map_err(|err| RuntimeError::Schema(err.to_string()))
523        })
524    }
525}
526
527pub trait PostgresProviderExt {
528    fn use_postgres_provider(&mut self, executor: PgMutationExecutor) -> &mut Self;
529}
530
531impl PostgresProviderExt for UserContext {
532    fn use_postgres_provider(&mut self, executor: PgMutationExecutor) -> &mut Self {
533        self.insert_resource(PostgresDialect);
534        self.insert_resource(executor);
535        self.set_schema_provider(PostgresSchemaProvider);
536        self
537    }
538}
539
540#[derive(Clone)]
541pub struct PgIdSpaceGenerator {
542    pool: Pool,
543    table_name: String,
544}
545
546impl PgIdSpaceGenerator {
547    pub fn new(pool: Pool) -> Self {
548        Self {
549            pool,
550            table_name: DEFAULT_ID_SPACE_TABLE.to_owned(),
551        }
552    }
553
554    pub fn from_executor(executor: PgMutationExecutor) -> Self {
555        Self::new(executor.pool())
556    }
557
558    pub fn with_table_name(mut self, table_name: impl Into<String>) -> Self {
559        self.table_name = table_name.into();
560        self
561    }
562
563    pub async fn ensure_table(&self) -> Result<(), MutationExecutorError> {
564        PgMutationExecutor::new(self.pool.clone())
565            .ensure_id_space_table(&self.table_name)
566            .await
567    }
568
569    pub async fn next_id(&self, entity: &str) -> Result<u64, MutationExecutorError> {
570        self.ensure_table().await?;
571        let update_sql = format!(
572            "UPDATE {} SET current_level = current_level + 1 WHERE type_name = $1 RETURNING current_level",
573            quote_ident(&self.table_name)
574        );
575        let client = self
576            .pool
577            .get()
578            .await
579            .map_err(|e| MutationExecutorError::Pool(e.to_string()))?;
580        let row = client.query_opt(&update_sql, &[&entity]).await?;
581
582        let id = match row {
583            Some(r) => {
584                let level: i64 = r.try_get(0)?;
585                level
586            }
587            None => {
588                let insert_sql = format!(
589                    "INSERT INTO {} (type_name, current_level) VALUES ($1, 1) RETURNING current_level",
590                    quote_ident(&self.table_name)
591                );
592                let insert_res = client.query_one(&insert_sql, &[&entity]).await;
593                match insert_res {
594                    Ok(r) => {
595                        let level: i64 = r.try_get(0)?;
596                        level
597                    }
598                    Err(_) => {
599                        let row = client.query_one(&update_sql, &[&entity]).await?;
600                        let level: i64 = row.try_get(0)?;
601                        level
602                    }
603                }
604            }
605        };
606
607        u64::try_from(id).map_err(|_| {
608            MutationExecutorError::Bind(format!("generated id {id} cannot be represented as u64"))
609        })
610    }
611}
612
613impl InternalIdGenerator for PgIdSpaceGenerator {
614    fn generate_id(&self, entity: &str) -> Result<u64, RuntimeError> {
615        let generator = self.clone();
616        let entity = entity.to_owned();
617        block_on_id_generation(async move { generator.next_id(&entity).await })
618    }
619}
620
621fn block_on_id_generation<F>(future: F) -> Result<u64, RuntimeError>
622where
623    F: Future<Output = Result<u64, MutationExecutorError>> + Send + 'static,
624{
625    let result = match tokio::runtime::Handle::try_current() {
626        Ok(handle) => tokio::task::block_in_place(|| handle.block_on(future)),
627        Err(_) => tokio::runtime::Builder::new_current_thread()
628            .enable_all()
629            .build()
630            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))?
631            .block_on(future),
632    };
633    result.map_err(|err| RuntimeError::IdGeneration(err.to_string()))
634}
635
636fn quote_ident(ident: &str) -> String {
637    quote_identifier_if_needed(ident, '"')
638}
639
640/// Strip wrapping identifier quotes from a SQL identifier so that bare column
641/// names returned by `information_schema.columns` can be compared with
642/// potentially-quoted `PropertyDescriptor::column_name` values.
643fn strip_identifier_quotes(ident: &str) -> &str {
644    let bytes = ident.as_bytes();
645    if bytes.len() >= 2 {
646        let (first, last) = (bytes[0], bytes[bytes.len() - 1]);
647        if (first == b'"' && last == b'"')
648            || (first == b'`' && last == b'`')
649            || (first == b'[' && last == b']')
650        {
651            return &ident[1..ident.len() - 1];
652        }
653    }
654    ident
655}
656
657fn try_parse_datetime_from_str(s: &str) -> Option<chrono::DateTime<chrono::Utc>> {
658    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
659        return Some(dt.with_timezone(&chrono::Utc));
660    }
661    if let Ok(ndt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
662        return Some(chrono::DateTime::from_naive_utc_and_offset(
663            ndt,
664            chrono::Utc,
665        ));
666    }
667    if let Ok(nd) = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") {
668        let ndt = nd.and_hms_opt(0, 0, 0)?;
669        return Some(chrono::DateTime::from_naive_utc_and_offset(
670            ndt,
671            chrono::Utc,
672        ));
673    }
674    None
675}
676
677struct PgArgs {
678    values: Vec<Box<dyn tokio_postgres::types::ToSql + Sync + Send>>,
679}
680impl PgArgs {
681    fn add<T: tokio_postgres::types::ToSql + Sync + Send + 'static>(&mut self, v: T) {
682        self.values.push(Box::new(v));
683    }
684    fn as_refs(&self) -> Vec<&(dyn tokio_postgres::types::ToSql + Sync)> {
685        self.values.iter().map(|b| b.as_ref() as _).collect()
686    }
687}
688
689fn bind_pg(args: &mut PgArgs, value: &Value) -> Result<(), MutationExecutorError> {
690    match value {
691        Value::Null => {
692            args.add(Option::<i32>::None);
693        }
694        Value::Bool(v) => args.add(*v),
695        Value::I64(v) => args.add(*v),
696        Value::U64(v) => {
697            let v = i64::try_from(*v).map_err(|_| {
698                MutationExecutorError::Bind(format!("u64 value {v} exceeds i64 range"))
699            })?;
700            args.add(v);
701        }
702        Value::F64(v) => args.add(*v),
703        Value::Decimal(v) => args.add(*v),
704        Value::Text(v) => match try_parse_datetime_from_str(v) {
705            Some(dt) => args.add(dt),
706            None => args.add(v.clone()),
707        },
708        Value::Json(v) => {
709            let j_val: serde_json::Value =
710                serde_json::to_value(v).map_err(|e| MutationExecutorError::Bind(e.to_string()))?;
711            args.add(j_val);
712        }
713        Value::Date(v) => args.add(*v),
714        Value::Timestamp(v) => args.add(*v),
715        Value::Object(_) => return Err(MutationExecutorError::UnsupportedValue("object")),
716        Value::List(values) => bind_pg_list(args, values)?,
717    }
718    Ok(())
719}
720
721fn bind_pg_list(args: &mut PgArgs, values: &[Value]) -> Result<(), MutationExecutorError> {
722    let Some(first) = values.first() else {
723        return Err(MutationExecutorError::UnsupportedValue("empty list"));
724    };
725    match first {
726        Value::Bool(_) => {
727            let values = values
728                .iter()
729                .map(|value| match value {
730                    Value::Bool(value) => Ok(*value),
731                    _ => Err(MutationExecutorError::UnsupportedValue("mixed bool list")),
732                })
733                .collect::<Result<Vec<_>, _>>()?;
734            args.add(values);
735        }
736        Value::I64(_) => {
737            let values = values
738                .iter()
739                .map(|value| match value {
740                    Value::I64(value) => Ok(*value),
741                    _ => Err(MutationExecutorError::UnsupportedValue("mixed i64 list")),
742                })
743                .collect::<Result<Vec<_>, _>>()?;
744            args.add(values);
745        }
746        Value::U64(_) => {
747            let values = values
748                .iter()
749                .map(|value| match value {
750                    Value::U64(value) => i64::try_from(*value).map_err(|_| {
751                        MutationExecutorError::Bind(format!("u64 value {value} exceeds i64 range"))
752                    }),
753                    _ => Err(MutationExecutorError::UnsupportedValue("mixed u64 list")),
754                })
755                .collect::<Result<Vec<_>, _>>()?;
756            args.add(values);
757        }
758        Value::F64(_) => {
759            let values = values
760                .iter()
761                .map(|value| match value {
762                    Value::F64(value) => Ok(*value),
763                    _ => Err(MutationExecutorError::UnsupportedValue("mixed f64 list")),
764                })
765                .collect::<Result<Vec<_>, _>>()?;
766            args.add(values);
767        }
768        Value::Decimal(_) => {
769            let values = values
770                .iter()
771                .map(|value| match value {
772                    Value::Decimal(value) => Ok(*value),
773                    _ => Err(MutationExecutorError::UnsupportedValue(
774                        "mixed decimal list",
775                    )),
776                })
777                .collect::<Result<Vec<_>, _>>()?;
778            args.add(values);
779        }
780        Value::Text(_) => {
781            let values = values
782                .iter()
783                .map(|value| match value {
784                    Value::Text(value) => Ok(value.clone()),
785                    _ => Err(MutationExecutorError::UnsupportedValue("mixed text list")),
786                })
787                .collect::<Result<Vec<_>, _>>()?;
788            args.add(values);
789        }
790        Value::Date(_) => {
791            let values = values
792                .iter()
793                .map(|value| match value {
794                    Value::Date(value) => Ok(*value),
795                    _ => Err(MutationExecutorError::UnsupportedValue("mixed date list")),
796                })
797                .collect::<Result<Vec<_>, _>>()?;
798            args.add(values);
799        }
800        Value::Timestamp(_) => {
801            let values = values
802                .iter()
803                .map(|value| match value {
804                    Value::Timestamp(value) => Ok(*value),
805                    _ => Err(MutationExecutorError::UnsupportedValue(
806                        "mixed timestamp list",
807                    )),
808                })
809                .collect::<Result<Vec<_>, _>>()?;
810            args.add(values);
811        }
812        Value::Null => return Err(MutationExecutorError::UnsupportedValue("null list")),
813        Value::Json(_) => return Err(MutationExecutorError::UnsupportedValue("json list")),
814        Value::Object(_) => return Err(MutationExecutorError::UnsupportedValue("object list")),
815        Value::List(_) => return Err(MutationExecutorError::UnsupportedValue("nested list")),
816    }
817    Ok(())
818}
819
820fn decode_pg_row(row: &tokio_postgres::Row) -> Result<Record, MutationExecutorError> {
821    let mut record = BTreeMap::new();
822    for (index, column) in row.columns().iter().enumerate() {
823        let name = column.name().to_owned();
824        let type_name = column.type_().name().to_ascii_uppercase();
825
826        let value = match type_name.as_str() {
827            "BOOL" | "BOOLEAN" => {
828                let v: Option<bool> = row.try_get(index)?;
829                match v {
830                    Some(v) => Value::Bool(v),
831                    None => Value::Null,
832                }
833            }
834            "INT2" => {
835                let v: Option<i16> = row.try_get(index)?;
836                match v {
837                    Some(v) => Value::I64(v as i64),
838                    None => Value::Null,
839                }
840            }
841            "INT4" => {
842                let v: Option<i32> = row.try_get(index)?;
843                match v {
844                    Some(v) => Value::I64(v as i64),
845                    None => Value::Null,
846                }
847            }
848            "INT8" => {
849                let v: Option<i64> = row.try_get(index)?;
850                match v {
851                    Some(v) => Value::I64(v),
852                    None => Value::Null,
853                }
854            }
855            "FLOAT4" => {
856                let v: Option<f32> = row.try_get(index)?;
857                match v {
858                    Some(v) => Value::F64(v as f64),
859                    None => Value::Null,
860                }
861            }
862            "FLOAT8" => {
863                let v: Option<f64> = row.try_get(index)?;
864                match v {
865                    Some(v) => Value::F64(v),
866                    None => Value::Null,
867                }
868            }
869            "NUMERIC" => {
870                let v: Option<Decimal> = row.try_get(index)?;
871                match v {
872                    Some(v) => Value::Decimal(v),
873                    None => Value::Null,
874                }
875            }
876            "JSON" | "JSONB" => {
877                let v: Option<serde_json::Value> = row.try_get(index)?;
878                match v {
879                    Some(j) => Value::Json(j.into()),
880                    None => Value::Null,
881                }
882            }
883            "DATE" => {
884                let v: Option<NaiveDate> = row.try_get(index)?;
885                match v {
886                    Some(v) => Value::Date(v),
887                    None => Value::Null,
888                }
889            }
890            "TIMESTAMP" | "TIMESTAMPTZ" => {
891                let v: Option<DateTime<Utc>> = row.try_get(index)?;
892                match v {
893                    Some(v) => Value::Timestamp(v),
894                    None => Value::Null,
895                }
896            }
897            "TEXT" | "VARCHAR" | "BPCHAR" | "NAME" | "UUID" => {
898                let v: Option<String> = row.try_get(index)?;
899                match v {
900                    Some(v) => Value::Text(v),
901                    None => Value::Null,
902                }
903            }
904            other => {
905                return Err(MutationExecutorError::UnsupportedColumnType(
906                    other.to_owned(),
907                ));
908            }
909        };
910        record.insert(name, value);
911    }
912    Ok(record)
913}
914
915#[cfg(test)]
916mod tests {
917    use super::*;
918    use teaql_core::{DeleteCommand, RecoverCommand};
919
920    fn entity() -> EntityDescriptor {
921        EntityDescriptor::new("Order")
922            .table_name("orders")
923            .property(
924                PropertyDescriptor::new("id", DataType::U64)
925                    .column_name("id")
926                    .id()
927                    .not_null(),
928            )
929            .property(
930                PropertyDescriptor::new("version", DataType::I64)
931                    .column_name("version")
932                    .version()
933                    .not_null(),
934            )
935            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
936    }
937
938    #[test]
939    fn postgres_dialect_compiles_mutations_with_numbered_placeholders() {
940        let insert = PostgresDialect
941            .compile_insert(
942                &entity(),
943                &InsertCommand::new("Order")
944                    .value("id", 1_u64)
945                    .value("name", "A"),
946            )
947            .unwrap();
948        assert_eq!(insert.sql, "INSERT INTO orders (id, name) VALUES ($1, $2)");
949
950        let update = PostgresDialect
951            .compile_update(
952                &entity(),
953                &UpdateCommand::new("Order", 1_u64)
954                    .expected_version(3)
955                    .value("name", "B"),
956            )
957            .unwrap();
958        assert_eq!(
959            update.sql,
960            "UPDATE orders SET name = $1, version = $2 WHERE id = $3 AND version = $4"
961        );
962
963        let delete = PostgresDialect
964            .compile_delete(
965                &entity(),
966                &DeleteCommand::new("Order", 1_u64).expected_version(3),
967            )
968            .unwrap();
969        let recover = PostgresDialect
970            .compile_recover(&entity(), &RecoverCommand::new("Order", 1_u64, -4))
971            .unwrap();
972        assert_eq!(
973            delete.sql,
974            "UPDATE orders SET version = $1 WHERE id = $2 AND version = $3"
975        );
976        assert_eq!(
977            recover.sql,
978            "UPDATE orders SET version = $1 WHERE id = $2 AND version = $3"
979        );
980    }
981
982    #[test]
983    fn postgres_dialect_compiles_schema_and_large_in_array_binds() {
984        let create = PostgresDialect.compile_create_table(&entity()).unwrap();
985        assert_eq!(
986            create,
987            "CREATE TABLE IF NOT EXISTS orders (id BIGINT PRIMARY KEY NOT NULL, version BIGINT NOT NULL, name TEXT)"
988        );
989        assert!(
990            PostgresDialect
991                .schema_setup_sqls()
992                .iter()
993                .any(|sql| sql.contains("CREATE OR REPLACE FUNCTION soundex"))
994        );
995
996        let query = PostgresDialect
997            .compile_select(
998                &entity(),
999                &SelectQuery::new("Order")
1000                    .filter(Expr::in_large(
1001                        "id",
1002                        vec![Value::from(1_u64), Value::from(2_u64)],
1003                    ))
1004                    .order_asc("id"),
1005            )
1006            .unwrap();
1007        assert_eq!(
1008            query.sql,
1009            "SELECT id, version, name FROM orders WHERE (id = ANY($1)) ORDER BY id ASC"
1010        );
1011        assert_eq!(
1012            query.params,
1013            vec![Value::List(vec![Value::U64(1), Value::U64(2)])]
1014        );
1015    }
1016}
teaql_provider_postgres/lib.rs

teaql_provider_postgres/
lib.rs
