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        } else {
423            let query = compile_initial_graph_insert(dialect, entity, graph)?;
424            executor.execute(&query).await?;
425        }
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 = if tokio::runtime::Handle::try_current().is_ok() {
626        let handle = tokio::runtime::Handle::current();
627        tokio::task::block_in_place(|| handle.block_on(future))
628    } else {
629        tokio::runtime::Builder::new_current_thread()
630            .enable_all()
631            .build()
632            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))?
633            .block_on(future)
634    };
635    result.map_err(|err| RuntimeError::IdGeneration(err.to_string()))
636}
637
638fn quote_ident(ident: &str) -> String {
639    quote_identifier_if_needed(ident, '"')
640}
641
642/// Strip wrapping identifier quotes from a SQL identifier so that bare column
643/// names returned by `information_schema.columns` can be compared with
644/// potentially-quoted `PropertyDescriptor::column_name` values.
645fn strip_identifier_quotes(ident: &str) -> &str {
646    let bytes = ident.as_bytes();
647    if bytes.len() >= 2 {
648        let (first, last) = (bytes[0], bytes[bytes.len() - 1]);
649        if (first == b'"' && last == b'"')
650            || (first == b'`' && last == b'`')
651            || (first == b'[' && last == b']')
652        {
653            return &ident[1..ident.len() - 1];
654        }
655    }
656    ident
657}
658
659fn try_parse_datetime_from_str(s: &str) -> Option<chrono::DateTime<chrono::Utc>> {
660    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
661        return Some(dt.with_timezone(&chrono::Utc));
662    }
663    if let Ok(ndt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
664        return Some(chrono::DateTime::from_naive_utc_and_offset(
665            ndt,
666            chrono::Utc,
667        ));
668    }
669    if let Ok(nd) = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") {
670        let ndt = nd.and_hms_opt(0, 0, 0)?;
671        return Some(chrono::DateTime::from_naive_utc_and_offset(
672            ndt,
673            chrono::Utc,
674        ));
675    }
676    None
677}
678
679struct PgArgs {
680    values: Vec<Box<dyn tokio_postgres::types::ToSql + Sync + Send>>,
681}
682impl PgArgs {
683    fn add<T: tokio_postgres::types::ToSql + Sync + Send + 'static>(&mut self, v: T) {
684        self.values.push(Box::new(v));
685    }
686    fn as_refs(&self) -> Vec<&(dyn tokio_postgres::types::ToSql + Sync)> {
687        self.values.iter().map(|b| b.as_ref() as _).collect()
688    }
689}
690
691fn bind_pg(args: &mut PgArgs, value: &Value) -> Result<(), MutationExecutorError> {
692    match value {
693        Value::Null => {
694            args.add(Option::<i32>::None);
695        }
696        Value::Bool(v) => args.add(*v),
697        Value::I64(v) => args.add(*v),
698        Value::U64(v) => {
699            let v = i64::try_from(*v).map_err(|_| {
700                MutationExecutorError::Bind(format!("u64 value {v} exceeds i64 range"))
701            })?;
702            args.add(v);
703        }
704        Value::F64(v) => args.add(*v),
705        Value::Decimal(v) => args.add(*v),
706        Value::Text(v) => {
707            if let Some(dt) = try_parse_datetime_from_str(v) {
708                args.add(dt);
709            } else {
710                args.add(v.clone());
711            }
712        }
713        Value::Json(v) => {
714            let j_val: serde_json::Value =
715                serde_json::to_value(v).map_err(|e| MutationExecutorError::Bind(e.to_string()))?;
716            args.add(j_val);
717        }
718        Value::Date(v) => args.add(*v),
719        Value::Timestamp(v) => args.add(*v),
720        Value::Object(_) => return Err(MutationExecutorError::UnsupportedValue("object")),
721        Value::List(values) => bind_pg_list(args, values)?,
722    }
723    Ok(())
724}
725
726fn bind_pg_list(args: &mut PgArgs, values: &[Value]) -> Result<(), MutationExecutorError> {
727    let Some(first) = values.first() else {
728        return Err(MutationExecutorError::UnsupportedValue("empty list"));
729    };
730    match first {
731        Value::Bool(_) => {
732            let values = values
733                .iter()
734                .map(|value| match value {
735                    Value::Bool(value) => Ok(*value),
736                    _ => Err(MutationExecutorError::UnsupportedValue("mixed bool list")),
737                })
738                .collect::<Result<Vec<_>, _>>()?;
739            args.add(values);
740        }
741        Value::I64(_) => {
742            let values = values
743                .iter()
744                .map(|value| match value {
745                    Value::I64(value) => Ok(*value),
746                    _ => Err(MutationExecutorError::UnsupportedValue("mixed i64 list")),
747                })
748                .collect::<Result<Vec<_>, _>>()?;
749            args.add(values);
750        }
751        Value::U64(_) => {
752            let values = values
753                .iter()
754                .map(|value| match value {
755                    Value::U64(value) => i64::try_from(*value).map_err(|_| {
756                        MutationExecutorError::Bind(format!("u64 value {value} exceeds i64 range"))
757                    }),
758                    _ => Err(MutationExecutorError::UnsupportedValue("mixed u64 list")),
759                })
760                .collect::<Result<Vec<_>, _>>()?;
761            args.add(values);
762        }
763        Value::F64(_) => {
764            let values = values
765                .iter()
766                .map(|value| match value {
767                    Value::F64(value) => Ok(*value),
768                    _ => Err(MutationExecutorError::UnsupportedValue("mixed f64 list")),
769                })
770                .collect::<Result<Vec<_>, _>>()?;
771            args.add(values);
772        }
773        Value::Decimal(_) => {
774            let values = values
775                .iter()
776                .map(|value| match value {
777                    Value::Decimal(value) => Ok(*value),
778                    _ => Err(MutationExecutorError::UnsupportedValue(
779                        "mixed decimal list",
780                    )),
781                })
782                .collect::<Result<Vec<_>, _>>()?;
783            args.add(values);
784        }
785        Value::Text(_) => {
786            let values = values
787                .iter()
788                .map(|value| match value {
789                    Value::Text(value) => Ok(value.clone()),
790                    _ => Err(MutationExecutorError::UnsupportedValue("mixed text list")),
791                })
792                .collect::<Result<Vec<_>, _>>()?;
793            args.add(values);
794        }
795        Value::Date(_) => {
796            let values = values
797                .iter()
798                .map(|value| match value {
799                    Value::Date(value) => Ok(*value),
800                    _ => Err(MutationExecutorError::UnsupportedValue("mixed date list")),
801                })
802                .collect::<Result<Vec<_>, _>>()?;
803            args.add(values);
804        }
805        Value::Timestamp(_) => {
806            let values = values
807                .iter()
808                .map(|value| match value {
809                    Value::Timestamp(value) => Ok(*value),
810                    _ => Err(MutationExecutorError::UnsupportedValue(
811                        "mixed timestamp list",
812                    )),
813                })
814                .collect::<Result<Vec<_>, _>>()?;
815            args.add(values);
816        }
817        Value::Null => return Err(MutationExecutorError::UnsupportedValue("null list")),
818        Value::Json(_) => return Err(MutationExecutorError::UnsupportedValue("json list")),
819        Value::Object(_) => return Err(MutationExecutorError::UnsupportedValue("object list")),
820        Value::List(_) => return Err(MutationExecutorError::UnsupportedValue("nested list")),
821    }
822    Ok(())
823}
824
825fn decode_pg_row(row: &tokio_postgres::Row) -> Result<Record, MutationExecutorError> {
826    let mut record = BTreeMap::new();
827    for (index, column) in row.columns().iter().enumerate() {
828        let name = column.name().to_owned();
829        let type_name = column.type_().name().to_ascii_uppercase();
830
831        let value = match type_name.as_str() {
832            "BOOL" | "BOOLEAN" => {
833                let v: Option<bool> = row.try_get(index)?;
834                match v {
835                    Some(v) => Value::Bool(v),
836                    None => Value::Null,
837                }
838            }
839            "INT2" => {
840                let v: Option<i16> = row.try_get(index)?;
841                match v {
842                    Some(v) => Value::I64(v as i64),
843                    None => Value::Null,
844                }
845            }
846            "INT4" => {
847                let v: Option<i32> = row.try_get(index)?;
848                match v {
849                    Some(v) => Value::I64(v as i64),
850                    None => Value::Null,
851                }
852            }
853            "INT8" => {
854                let v: Option<i64> = row.try_get(index)?;
855                match v {
856                    Some(v) => Value::I64(v),
857                    None => Value::Null,
858                }
859            }
860            "FLOAT4" => {
861                let v: Option<f32> = row.try_get(index)?;
862                match v {
863                    Some(v) => Value::F64(v as f64),
864                    None => Value::Null,
865                }
866            }
867            "FLOAT8" => {
868                let v: Option<f64> = row.try_get(index)?;
869                match v {
870                    Some(v) => Value::F64(v),
871                    None => Value::Null,
872                }
873            }
874            "NUMERIC" => {
875                let v: Option<Decimal> = row.try_get(index)?;
876                match v {
877                    Some(v) => Value::Decimal(v),
878                    None => Value::Null,
879                }
880            }
881            "JSON" | "JSONB" => {
882                let v: Option<serde_json::Value> = row.try_get(index)?;
883                match v {
884                    Some(j) => Value::Json(j.into()),
885                    None => Value::Null,
886                }
887            }
888            "DATE" => {
889                let v: Option<NaiveDate> = row.try_get(index)?;
890                match v {
891                    Some(v) => Value::Date(v),
892                    None => Value::Null,
893                }
894            }
895            "TIMESTAMP" | "TIMESTAMPTZ" => {
896                let v: Option<DateTime<Utc>> = row.try_get(index)?;
897                match v {
898                    Some(v) => Value::Timestamp(v),
899                    None => Value::Null,
900                }
901            }
902            "TEXT" | "VARCHAR" | "BPCHAR" | "NAME" | "UUID" => {
903                let v: Option<String> = row.try_get(index)?;
904                match v {
905                    Some(v) => Value::Text(v),
906                    None => Value::Null,
907                }
908            }
909            other => {
910                return Err(MutationExecutorError::UnsupportedColumnType(
911                    other.to_owned(),
912                ));
913            }
914        };
915        record.insert(name, value);
916    }
917    Ok(record)
918}
919
920#[cfg(test)]
921mod tests {
922    use super::*;
923    use teaql_core::{DeleteCommand, RecoverCommand};
924
925    fn entity() -> EntityDescriptor {
926        EntityDescriptor::new("Order")
927            .table_name("orders")
928            .property(
929                PropertyDescriptor::new("id", DataType::U64)
930                    .column_name("id")
931                    .id()
932                    .not_null(),
933            )
934            .property(
935                PropertyDescriptor::new("version", DataType::I64)
936                    .column_name("version")
937                    .version()
938                    .not_null(),
939            )
940            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
941    }
942
943    #[test]
944    fn postgres_dialect_compiles_mutations_with_numbered_placeholders() {
945        let insert = PostgresDialect
946            .compile_insert(
947                &entity(),
948                &InsertCommand::new("Order")
949                    .value("id", 1_u64)
950                    .value("name", "A"),
951            )
952            .unwrap();
953        assert_eq!(insert.sql, "INSERT INTO orders (id, name) VALUES ($1, $2)");
954
955        let update = PostgresDialect
956            .compile_update(
957                &entity(),
958                &UpdateCommand::new("Order", 1_u64)
959                    .expected_version(3)
960                    .value("name", "B"),
961            )
962            .unwrap();
963        assert_eq!(
964            update.sql,
965            "UPDATE orders SET name = $1, version = $2 WHERE id = $3 AND version = $4"
966        );
967
968        let delete = PostgresDialect
969            .compile_delete(
970                &entity(),
971                &DeleteCommand::new("Order", 1_u64).expected_version(3),
972            )
973            .unwrap();
974        let recover = PostgresDialect
975            .compile_recover(&entity(), &RecoverCommand::new("Order", 1_u64, -4))
976            .unwrap();
977        assert_eq!(
978            delete.sql,
979            "UPDATE orders SET version = $1 WHERE id = $2 AND version = $3"
980        );
981        assert_eq!(
982            recover.sql,
983            "UPDATE orders SET version = $1 WHERE id = $2 AND version = $3"
984        );
985    }
986
987    #[test]
988    fn postgres_dialect_compiles_schema_and_large_in_array_binds() {
989        let create = PostgresDialect.compile_create_table(&entity()).unwrap();
990        assert_eq!(
991            create,
992            "CREATE TABLE IF NOT EXISTS orders (id BIGINT PRIMARY KEY NOT NULL, version BIGINT NOT NULL, name TEXT)"
993        );
994        assert!(
995            PostgresDialect
996                .schema_setup_sqls()
997                .iter()
998                .any(|sql| sql.contains("CREATE OR REPLACE FUNCTION soundex"))
999        );
1000
1001        let query = PostgresDialect
1002            .compile_select(
1003                &entity(),
1004                &SelectQuery::new("Order")
1005                    .filter(Expr::in_large(
1006                        "id",
1007                        vec![Value::from(1_u64), Value::from(2_u64)],
1008                    ))
1009                    .order_asc("id"),
1010            )
1011            .unwrap();
1012        assert_eq!(
1013            query.sql,
1014            "SELECT id, version, name FROM orders WHERE (id = ANY($1)) ORDER BY id ASC"
1015        );
1016        assert_eq!(
1017            query.params,
1018            vec![Value::List(vec![Value::U64(1), Value::U64(2)])]
1019        );
1020    }
1021}
teaql_provider_postgres/lib.rs

teaql_provider_postgres/
lib.rs
