teaql_sql/
dialect.rs

1use teaql_core::{
2    Aggregate, AggregateFunction, BinaryOp, DataType, DeleteCommand, EntityDescriptor, Expr,
3    ExprFunction, OrderBy, PropertyDescriptor, RecoverCommand, SelectQuery, SortDirection, Value,
4};
5
6use crate::{CompiledQuery, DatabaseKind, SqlCompileError};
7
8const SQL_KEYWORDS: &[&str] = &[
9    "all", "alter", "and", "as", "asc", "between", "by", "case", "create", "delete", "desc",
10    "distinct", "drop", "exists", "false", "from", "group", "having", "in", "insert", "into", "is",
11    "join", "like", "limit", "not", "null", "offset", "on", "or", "order", "select", "set",
12    "table", "true", "type", "union", "update", "values", "where",
13];
14
15pub fn quote_identifier_if_needed(ident: &str, quote: char) -> String {
16    if is_wrapped_identifier(ident) {
17        return ident.to_owned();
18    }
19    if needs_quoted_identifier(ident) {
20        let quote_string = quote.to_string();
21        let escaped = ident.replace(quote, &(quote_string.clone() + &quote_string));
22        return format!("{quote}{escaped}{quote}");
23    }
24    ident.to_owned()
25}
26
27fn is_wrapped_identifier(ident: &str) -> bool {
28    (ident.starts_with('"') && ident.ends_with('"'))
29        || (ident.starts_with('`') && ident.ends_with('`'))
30        || (ident.starts_with('[') && ident.ends_with(']'))
31}
32
33fn needs_quoted_identifier(ident: &str) -> bool {
34    if ident.is_empty()
35        || SQL_KEYWORDS
36            .binary_search(&ident.to_ascii_lowercase().as_str())
37            .is_ok()
38    {
39        return true;
40    }
41    let mut chars = ident.chars();
42    match chars.next() {
43        Some(first) if first == '_' || first.is_ascii_alphabetic() => {}
44        _ => return true,
45    }
46    chars.any(|ch| ch != '_' && !ch.is_ascii_alphanumeric())
47}
48
49pub trait SqlDialect {
50    fn kind(&self) -> DatabaseKind;
51    fn quote_ident(&self, ident: &str) -> String;
52    fn placeholder(&self, index: usize) -> String;
53
54    /// Whether `IN_LARGE` / `NOT_IN_LARGE` bind the complete value list as a
55    /// single dialect-native array parameter instead of scalar placeholders.
56    fn large_in_uses_array_param(&self) -> bool {
57        false
58    }
59
60    fn prefers_small_parent_relation_probes(&self) -> bool {
61        false
62    }
63
64    fn schema_setup_sqls(&self) -> &'static [&'static str] {
65        &[]
66    }
67
68    fn schema_type_sql(
69        &self,
70        data_type: DataType,
71        _property: &PropertyDescriptor,
72    ) -> Result<&'static str, SqlCompileError> {
73        match data_type {
74            DataType::Bool => Ok("BOOLEAN"),
75            DataType::I64 | DataType::U64 => Ok("INTEGER"),
76            DataType::F64 => Ok("REAL"),
77            DataType::Decimal => Ok("NUMERIC"),
78            DataType::Text => Ok("VARCHAR(255)"),
79            DataType::LargeText | DataType::Json | DataType::Date | DataType::Timestamp => {
80                Ok("TEXT")
81            }
82        }
83    }
84
85    fn column_definition_sql(
86        &self,
87        property: &PropertyDescriptor,
88    ) -> Result<String, SqlCompileError> {
89        let mut parts = vec![
90            self.quote_ident(&property.column_name),
91            self.schema_type_sql(property.data_type, property)?
92                .to_owned(),
93        ];
94
95        if property.is_id {
96            parts.push("PRIMARY KEY".to_owned());
97        }
98        if property.is_id || !property.nullable {
99            parts.push("NOT NULL".to_owned());
100        }
101
102        Ok(parts.join(" "))
103    }
104
105    fn compile_create_table(&self, entity: &EntityDescriptor) -> Result<String, SqlCompileError> {
106        let columns = entity
107            .properties
108            .iter()
109            .map(|property| self.column_definition_sql(property))
110            .collect::<Result<Vec<_>, _>>()?
111            .join(", ");
112        Ok(format!(
113            "CREATE TABLE IF NOT EXISTS {} ({columns})",
114            self.quote_ident(&entity.table_name)
115        ))
116    }
117
118    fn schema_indexes_sqls(
119        &self,
120        entity: &EntityDescriptor,
121    ) -> Result<Vec<String>, SqlCompileError> {
122        let mut sqls = Vec::new();
123        let table_name_upper = entity.table_name.to_uppercase();
124        let quoted_table = self.quote_ident(&entity.table_name);
125
126        if let Some(version_col) = entity.properties.iter().find(|p| p.is_version) {
127            let default_id = "id".to_string();
128            let id_col = entity
129                .properties
130                .iter()
131                .find(|p| p.is_id)
132                .map(|p| &p.column_name)
133                .unwrap_or(&default_id);
134            let idx_name = format!("PK_{}_ID_VERSION", table_name_upper);
135            sqls.push(format!(
136                "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {} ({}, {})",
137                self.quote_ident(&idx_name),
138                quoted_table,
139                self.quote_ident(id_col),
140                self.quote_ident(&version_col.column_name)
141            ));
142        }
143
144        for p in &entity.properties {
145            if p.name.ends_with("Id")
146                || p.name.ends_with("Time")
147                || p.name.ends_with("_time")
148                || p.name == "create_time"
149                || p.name == "update_time"
150            {
151                let idx_name = format!("IDX_{}_{}", table_name_upper, p.column_name.to_uppercase());
152                sqls.push(format!(
153                    "CREATE INDEX IF NOT EXISTS {} ON {} ({})",
154                    self.quote_ident(&idx_name),
155                    quoted_table,
156                    self.quote_ident(&p.column_name)
157                ));
158            }
159        }
160
161        Ok(sqls)
162    }
163
164    fn fallback_default_value_sql(&self, data_type: DataType) -> &'static str {
165        match data_type {
166            DataType::Bool => "FALSE",
167            DataType::I64 | DataType::U64 | DataType::F64 | DataType::Decimal => "0",
168            DataType::Text | DataType::LargeText => "''",
169            DataType::Json => "'{}'",
170            DataType::Date => "'1970-01-01'",
171            DataType::Timestamp => "'1970-01-01 00:00:00Z'",
172        }
173    }
174
175    fn compile_add_column(
176        &self,
177        entity: &EntityDescriptor,
178        property: &PropertyDescriptor,
179    ) -> Result<String, SqlCompileError> {
180        let mut def = self.column_definition_sql(property)?;
181        if !property.nullable && !property.is_id {
182            def.push_str(" DEFAULT ");
183            def.push_str(self.fallback_default_value_sql(property.data_type));
184        }
185        Ok(format!(
186            "ALTER TABLE {} ADD COLUMN {}",
187            self.quote_ident(&entity.table_name),
188            def
189        ))
190    }
191
192    fn compile_select(
193        &self,
194        entity: &EntityDescriptor,
195        query: &SelectQuery,
196    ) -> Result<CompiledQuery, SqlCompileError> {
197        let mut params = Vec::new();
198        let sql = self.compile_select_sql(entity, query, &mut params)?;
199        Ok(CompiledQuery {
200            sql,
201            params,
202            comment: query.comment.clone(),
203        })
204    }
205
206    fn compile_select_sql(
207        &self,
208        entity: &EntityDescriptor,
209        query: &SelectQuery,
210        params: &mut Vec<Value>,
211    ) -> Result<String, SqlCompileError> {
212        if let Some(raw_sql) = &query.raw_sql {
213            return Ok(raw_sql.clone());
214        }
215
216        let mut projection = self.compile_projection(entity, query, params)?;
217        let partitioned_slice = query.partition_by.as_deref().zip(query.slice);
218        if let Some((partition_by, _)) = partitioned_slice {
219            let partition_column = self.column_sql(entity, partition_by)?;
220            let window_order = if query.order_by.is_empty() {
221                String::new()
222            } else {
223                let order_by = query
224                    .order_by
225                    .iter()
226                    .map(|order| self.order_by_sql(entity, order, params))
227                    .collect::<Result<Vec<_>, _>>()?
228                    .join(", ");
229                format!(" ORDER BY {order_by}")
230            };
231            let rank = self.quote_ident(teaql_core::PARTITION_RANK_PROPERTY);
232            projection.push_str(&format!(
233                ", ROW_NUMBER() OVER (PARTITION BY {partition_column}{window_order}) AS {rank}"
234            ));
235        }
236
237        let mut sql = format!(
238            "SELECT {projection} FROM {}",
239            self.quote_ident(&entity.table_name)
240        );
241
242        let mut where_parts = Vec::new();
243        if let Some(filter) = &query.filter {
244            where_parts.push(self.compile_expr(entity, filter, params)?);
245        }
246
247        if let Some(search_text) = &query.search_with_text {
248            let mut or_parts = Vec::new();
249            let like_value = format!("%{}%", search_text);
250            for property in &entity.properties {
251                if property.data_type == teaql_core::DataType::Text
252                    || property.data_type == teaql_core::DataType::LargeText
253                {
254                    params.push(teaql_core::Value::from(like_value.clone()));
255                    or_parts.push(format!(
256                        "{} LIKE {}",
257                        self.quote_ident(&property.column_name),
258                        self.placeholder(params.len())
259                    ));
260                }
261            }
262            if !or_parts.is_empty() {
263                where_parts.push(format!("({})", or_parts.join(" OR ")));
264            }
265        }
266
267        where_parts.extend(query.raw_sql_search_criteria.iter().cloned());
268        if !where_parts.is_empty() {
269            sql.push_str(" WHERE ");
270            sql.push_str(&where_parts.join(" AND "));
271        }
272
273        if let Some((_, slice)) = partitioned_slice {
274            let rank = self.quote_ident(teaql_core::PARTITION_RANK_PROPERTY);
275            let alias = self.quote_ident("__teaql_partitioned");
276            let mut predicates = vec![format!("{rank} > {}", slice.offset)];
277            if let Some(limit) = slice.limit {
278                predicates.push(format!("{rank} <= {}", slice.offset.saturating_add(limit)));
279            }
280            return Ok(format!(
281                "SELECT * FROM ({sql}) AS {alias} WHERE {} ORDER BY {rank}",
282                predicates.join(" AND ")
283            ));
284        }
285
286        if !query.group_by.is_empty() {
287            let group_by = query
288                .group_by
289                .iter()
290                .map(|field| self.column_sql(entity, field))
291                .collect::<Result<Vec<_>, _>>()?
292                .join(", ");
293            sql.push_str(" GROUP BY ");
294            sql.push_str(&group_by);
295        }
296
297        if let Some(having) = &query.having {
298            let having_sql = self.compile_expr(entity, having, params)?;
299            sql.push_str(" HAVING ");
300            sql.push_str(&having_sql);
301        }
302
303        if !query.order_by.is_empty() {
304            let order_by = query
305                .order_by
306                .iter()
307                .map(|order| self.order_by_sql(entity, order, params))
308                .collect::<Result<Vec<_>, _>>()?
309                .join(", ");
310            sql.push_str(" ORDER BY ");
311            sql.push_str(&order_by);
312        }
313
314        if let Some(slice) = query.slice {
315            if let Some(limit) = slice.limit {
316                sql.push_str(&format!(" LIMIT {limit}"));
317            }
318            if slice.offset > 0 {
319                sql.push_str(&format!(" OFFSET {}", slice.offset));
320            }
321        }
322
323        Ok(sql)
324    }
325
326    fn compile_insert(
327        &self,
328        entity: &EntityDescriptor,
329        command: &teaql_core::InsertCommand,
330    ) -> Result<CompiledQuery, SqlCompileError> {
331        let mut columns = Vec::new();
332        let mut placeholders = Vec::new();
333        let mut params = Vec::new();
334
335        for property in &entity.properties {
336            if let Some(value) = command.values.get(&property.name) {
337                columns.push(self.quote_ident(&property.column_name));
338                let mut v = value.clone();
339                if let Value::Null = v {
340                    v = Value::TypedNull(property.data_type);
341                }
342                params.push(v);
343                placeholders.push(self.placeholder(params.len()));
344            }
345        }
346
347        if columns.is_empty() {
348            return Err(SqlCompileError::EmptyMutation("insert".to_owned()));
349        }
350
351        Ok(CompiledQuery {
352            sql: format!(
353                "INSERT INTO {} ({}) VALUES ({})",
354                self.quote_ident(&entity.table_name),
355                columns.join(", "),
356                placeholders.join(", ")
357            ),
358            params,
359            comment: None,
360        })
361    }
362
363    fn compile_batch_insert(
364        &self,
365        entity: &EntityDescriptor,
366        command: &teaql_core::BatchInsertCommand,
367    ) -> Result<CompiledQuery, SqlCompileError> {
368        if command.batch_values.is_empty() {
369            return Err(SqlCompileError::EmptyMutation("batch_insert".to_owned()));
370        }
371
372        let mut columns = Vec::new();
373        let first_record = &command.batch_values[0];
374
375        for property in &entity.properties {
376            if first_record.contains_key(&property.name) {
377                columns.push(property.clone());
378            }
379        }
380
381        if columns.is_empty() {
382            return Err(SqlCompileError::EmptyMutation("batch_insert".to_owned()));
383        }
384
385        let column_names: Vec<String> = columns
386            .iter()
387            .map(|p| self.quote_ident(&p.column_name))
388            .collect();
389        let mut params = Vec::new();
390        let mut values_clauses = Vec::new();
391
392        for record in &command.batch_values {
393            let mut row_placeholders = Vec::new();
394            for property in &columns {
395                let mut value = record
396                    .get(&property.name)
397                    .cloned()
398                    .unwrap_or(teaql_core::Value::Null);
399                if let Value::Null = value {
400                    value = Value::TypedNull(property.data_type);
401                }
402                params.push(value);
403                row_placeholders.push(self.placeholder(params.len()));
404            }
405            values_clauses.push(format!("({})", row_placeholders.join(", ")));
406        }
407
408        Ok(CompiledQuery {
409            sql: format!(
410                "INSERT INTO {} ({}) VALUES {}",
411                self.quote_ident(&entity.table_name),
412                column_names.join(", "),
413                values_clauses.join(", ")
414            ),
415            params,
416            comment: None,
417        })
418    }
419
420    fn compile_update(
421        &self,
422        entity: &EntityDescriptor,
423        command: &teaql_core::UpdateCommand,
424    ) -> Result<CompiledQuery, SqlCompileError> {
425        let id_property = entity
426            .id_property()
427            .ok_or_else(|| SqlCompileError::MissingIdProperty(entity.name.clone()))?;
428        let mut assignments = Vec::new();
429        let mut params = Vec::new();
430
431        for property in &entity.properties {
432            if property.is_id {
433                continue;
434            }
435            if property.is_version && command.expected_version.is_some() {
436                continue;
437            }
438            if let Some(value) = command.values.get(&property.name) {
439                let mut v = value.clone();
440                if let Value::Null = v {
441                    v = Value::TypedNull(property.data_type);
442                }
443                params.push(v);
444                assignments.push(format!(
445                    "{} = {}",
446                    self.quote_ident(&property.column_name),
447                    self.placeholder(params.len())
448                ));
449            }
450        }
451
452        if let Some(expected_version) = command.expected_version {
453            let version_property = entity
454                .version_property()
455                .ok_or_else(|| SqlCompileError::MissingVersionProperty(entity.name.clone()))?;
456            params.push(Value::I64(expected_version + 1));
457            assignments.push(format!(
458                "{} = {}",
459                self.quote_ident(&version_property.column_name),
460                self.placeholder(params.len())
461            ));
462        }
463
464        if assignments.is_empty() {
465            return Err(SqlCompileError::EmptyMutation("update".to_owned()));
466        }
467
468        params.push(command.id.clone());
469        let mut predicates = vec![format!(
470            "{} = {}",
471            self.quote_ident(&id_property.column_name),
472            self.placeholder(params.len())
473        )];
474
475        if let Some(expected_version) = command.expected_version {
476            let version_property = entity
477                .version_property()
478                .ok_or_else(|| SqlCompileError::MissingVersionProperty(entity.name.clone()))?;
479            params.push(Value::I64(expected_version));
480            predicates.push(format!(
481                "{} = {}",
482                self.quote_ident(&version_property.column_name),
483                self.placeholder(params.len())
484            ));
485        }
486
487        Ok(CompiledQuery {
488            sql: format!(
489                "UPDATE {} SET {} WHERE {}",
490                self.quote_ident(&entity.table_name),
491                assignments.join(", "),
492                predicates.join(" AND ")
493            ),
494            params,
495            comment: None,
496        })
497    }
498
499    fn compile_batch_update(
500        &self,
501        entity: &EntityDescriptor,
502        command: &teaql_core::BatchUpdateCommand,
503    ) -> Result<CompiledQuery, SqlCompileError> {
504        if command.batch_values.is_empty() {
505            return Err(SqlCompileError::EmptyMutation("batch_update".to_owned()));
506        }
507
508        let id_property = entity
509            .id_property()
510            .ok_or_else(|| SqlCompileError::MissingIdProperty(entity.name.clone()))?;
511
512        let mut params = Vec::new();
513        let mut set_clauses = Vec::new();
514
515        // Build CASE statement for each updated field
516        for field_name in &command.update_fields {
517            let property = entity
518                .property_by_name(field_name)
519                .ok_or_else(|| SqlCompileError::UnknownField(field_name.clone()))?;
520
521            let mut case_parts = Vec::new();
522            case_parts.push(format!(
523                "CASE {}",
524                self.quote_ident(&id_property.column_name)
525            ));
526
527            for (i, record) in command.batch_values.iter().enumerate() {
528                let id = &command.batch_ids[i];
529                let mut val = record
530                    .get(field_name)
531                    .cloned()
532                    .unwrap_or(teaql_core::Value::Null);
533                if let Value::Null = val {
534                    val = Value::TypedNull(property.data_type);
535                }
536
537                params.push(id.clone());
538                let id_ph = self.placeholder(params.len());
539
540                params.push(val);
541                let val_ph = self.placeholder(params.len());
542
543                case_parts.push(format!("WHEN {} THEN {}", id_ph, val_ph));
544            }
545
546            case_parts.push(format!(
547                "ELSE {} END",
548                self.quote_ident(&property.column_name)
549            ));
550            set_clauses.push(format!(
551                "{} = {}",
552                self.quote_ident(&property.column_name),
553                case_parts.join(" ")
554            ));
555        }
556
557        let mut has_versions = false;
558        if let Some(version_property) = entity.version_property() {
559            let mut case_parts = Vec::new();
560            case_parts.push(format!(
561                "CASE {}",
562                self.quote_ident(&id_property.column_name)
563            ));
564
565            for (i, exp_ver_opt) in command.batch_expected_versions.iter().enumerate() {
566                if let Some(exp_ver) = exp_ver_opt {
567                    has_versions = true;
568                    let id = &command.batch_ids[i];
569
570                    params.push(id.clone());
571                    let id_ph = self.placeholder(params.len());
572
573                    params.push(teaql_core::Value::I64(*exp_ver + 1));
574                    let val_ph = self.placeholder(params.len());
575
576                    case_parts.push(format!("WHEN {} THEN {}", id_ph, val_ph));
577                }
578            }
579
580            if has_versions {
581                case_parts.push(format!(
582                    "ELSE {} END",
583                    self.quote_ident(&version_property.column_name)
584                ));
585                set_clauses.push(format!(
586                    "{} = {}",
587                    self.quote_ident(&version_property.column_name),
588                    case_parts.join(" ")
589                ));
590            }
591        }
592
593        if set_clauses.is_empty() {
594            return Err(SqlCompileError::EmptyMutation("batch_update".to_owned()));
595        }
596
597        let mut in_placeholders = Vec::new();
598        for id in &command.batch_ids {
599            params.push(id.clone());
600            in_placeholders.push(self.placeholder(params.len()));
601        }
602        let mut predicates = vec![format!(
603            "{} IN ({})",
604            self.quote_ident(&id_property.column_name),
605            in_placeholders.join(", ")
606        )];
607
608        if has_versions {
609            let version_property = entity.version_property().unwrap();
610            let mut case_parts = Vec::new();
611            case_parts.push(format!(
612                "CASE {}",
613                self.quote_ident(&id_property.column_name)
614            ));
615
616            for (i, exp_ver_opt) in command.batch_expected_versions.iter().enumerate() {
617                if let Some(exp_ver) = exp_ver_opt {
618                    let id = &command.batch_ids[i];
619
620                    params.push(id.clone());
621                    let id_ph = self.placeholder(params.len());
622
623                    params.push(teaql_core::Value::I64(*exp_ver));
624                    let val_ph = self.placeholder(params.len());
625
626                    case_parts.push(format!("WHEN {} THEN {}", id_ph, val_ph));
627                }
628            }
629            case_parts.push(format!(
630                "ELSE {} END",
631                self.quote_ident(&version_property.column_name)
632            ));
633
634            predicates.push(format!(
635                "{} = {}",
636                self.quote_ident(&version_property.column_name),
637                case_parts.join(" ")
638            ));
639        }
640
641        Ok(CompiledQuery {
642            sql: format!(
643                "UPDATE {} SET {} WHERE {}",
644                self.quote_ident(&entity.table_name),
645                set_clauses.join(", "),
646                predicates.join(" AND ")
647            ),
648            params,
649            comment: None,
650        })
651    }
652
653    fn compile_delete(
654        &self,
655        entity: &EntityDescriptor,
656        command: &DeleteCommand,
657    ) -> Result<CompiledQuery, SqlCompileError> {
658        let id_property = entity
659            .id_property()
660            .ok_or_else(|| SqlCompileError::MissingIdProperty(entity.name.clone()))?;
661        let mut params = Vec::new();
662
663        if command.soft_delete {
664            let version_property = entity
665                .version_property()
666                .ok_or_else(|| SqlCompileError::MissingVersionProperty(entity.name.clone()))?;
667            params.push(match command.expected_version {
668                Some(version) => Value::I64(-(version + 1)),
669                None => Value::I64(-1),
670            });
671
672            params.push(command.id.clone());
673            let mut predicates = vec![format!(
674                "{} = {}",
675                self.quote_ident(&id_property.column_name),
676                self.placeholder(params.len())
677            )];
678
679            if let Some(expected_version) = command.expected_version {
680                params.push(Value::I64(expected_version));
681                predicates.push(format!(
682                    "{} = {}",
683                    self.quote_ident(&version_property.column_name),
684                    self.placeholder(params.len())
685                ));
686            }
687
688            return Ok(CompiledQuery {
689                sql: format!(
690                    "UPDATE {} SET {} = {} WHERE {}",
691                    self.quote_ident(&entity.table_name),
692                    self.quote_ident(&version_property.column_name),
693                    self.placeholder(1),
694                    predicates.join(" AND ")
695                ),
696                params,
697                comment: None,
698            });
699        }
700
701        params.push(command.id.clone());
702        let mut predicates = vec![format!(
703            "{} = {}",
704            self.quote_ident(&id_property.column_name),
705            self.placeholder(params.len())
706        )];
707
708        if let Some(expected_version) = command.expected_version {
709            let version_property = entity
710                .version_property()
711                .ok_or_else(|| SqlCompileError::MissingVersionProperty(entity.name.clone()))?;
712            params.push(Value::I64(expected_version));
713            predicates.push(format!(
714                "{} = {}",
715                self.quote_ident(&version_property.column_name),
716                self.placeholder(params.len())
717            ));
718        }
719
720        Ok(CompiledQuery {
721            sql: format!(
722                "DELETE FROM {} WHERE {}",
723                self.quote_ident(&entity.table_name),
724                predicates.join(" AND ")
725            ),
726            params,
727            comment: None,
728        })
729    }
730
731    fn compile_recover(
732        &self,
733        entity: &EntityDescriptor,
734        command: &RecoverCommand,
735    ) -> Result<CompiledQuery, SqlCompileError> {
736        if command.expected_version >= 0 {
737            return Err(SqlCompileError::InvalidRecoverVersion(
738                command.expected_version,
739            ));
740        }
741
742        let id_property = entity
743            .id_property()
744            .ok_or_else(|| SqlCompileError::MissingIdProperty(entity.name.clone()))?;
745        let version_property = entity
746            .version_property()
747            .ok_or_else(|| SqlCompileError::MissingVersionProperty(entity.name.clone()))?;
748        let params = vec![
749            Value::I64(-command.expected_version + 1),
750            command.id.clone(),
751            Value::I64(command.expected_version),
752        ];
753
754        Ok(CompiledQuery {
755            sql: format!(
756                "UPDATE {} SET {} = {} WHERE {} = {} AND {} = {}",
757                self.quote_ident(&entity.table_name),
758                self.quote_ident(&version_property.column_name),
759                self.placeholder(1),
760                self.quote_ident(&id_property.column_name),
761                self.placeholder(2),
762                self.quote_ident(&version_property.column_name),
763                self.placeholder(3),
764            ),
765            params,
766            comment: None,
767        })
768    }
769
770    fn column_sql(
771        &self,
772        entity: &EntityDescriptor,
773        field: &str,
774    ) -> Result<String, SqlCompileError> {
775        let property = entity
776            .property_by_name(field)
777            .ok_or_else(|| SqlCompileError::UnknownField(field.to_owned()))?;
778        Ok(self.quote_ident(&property.column_name))
779    }
780
781    fn order_by_sql(
782        &self,
783        entity: &EntityDescriptor,
784        order_by: &OrderBy,
785        params: &mut Vec<Value>,
786    ) -> Result<String, SqlCompileError> {
787        let field = self.resolve_order_field(entity, order_by, params)?;
788        let direction = match order_by.direction {
789            SortDirection::Asc => "ASC",
790            SortDirection::Desc => "DESC",
791        };
792        Ok(format!("{field} {direction}"))
793    }
794
795    fn select_projection(
796        &self,
797        entity: &EntityDescriptor,
798        query: &SelectQuery,
799        params: &mut Vec<Value>,
800    ) -> Result<String, SqlCompileError> {
801        let property_projection = |property: &PropertyDescriptor| self.column_with_alias(property);
802
803        if query.projection.is_empty()
804            && query.expr_projection.is_empty()
805            && query.raw_projections.is_empty()
806            && query.dynamic_properties.is_empty()
807        {
808            return Ok(entity
809                .properties
810                .iter()
811                .map(property_projection)
812                .collect::<Vec<_>>()
813                .join(", "));
814        }
815        // Generated relation selections may request an identity field both as
816        // a base projection and as part of the selected entity graph. MySQL
817        // rejects duplicate column names inside the partition/window derived
818        // table, so preserve first-seen order while removing duplicates.
819        let mut seen_fields = std::collections::BTreeSet::new();
820        let mut parts = Vec::new();
821        for field in &query.projection {
822            if !seen_fields.insert(field.as_str()) {
823                continue;
824            }
825            let property = entity
826                .property_by_name(field)
827                .ok_or_else(|| SqlCompileError::UnknownField(field.to_owned()))?;
828            parts.push(property_projection(property));
829        }
830        for projection in &query.expr_projection {
831            let expr = self.compile_expr(entity, &projection.expr, params)?;
832            parts.push(format!("{expr} AS {}", self.quote_ident(&projection.alias)));
833        }
834        for projection in query
835            .raw_projections
836            .iter()
837            .chain(query.dynamic_properties.iter())
838        {
839            parts.push(format!(
840                "{} AS {}",
841                projection.raw_sql_segment,
842                self.quote_ident(&projection.property_name)
843            ));
844        }
845        Ok(parts.join(", "))
846    }
847
848    fn aggregate_projection(
849        &self,
850        entity: &EntityDescriptor,
851        query: &SelectQuery,
852        params: &mut Vec<Value>,
853    ) -> Result<String, SqlCompileError> {
854        let mut parts = Vec::new();
855        // Aggregate queries must not inherit the entity's ordinary/default projection.
856        // Only grouping keys may be projected alongside aggregate expressions;
857        // otherwise generated requests produce invalid SQL such as
858        // `SELECT id, COUNT(id) ...` without grouping by `id`.
859        for field in &query.group_by {
860            let column = self.column_sql(entity, field)?;
861            if !parts.contains(&column) {
862                parts.push(column);
863            }
864        }
865        for projection in &query.expr_projection {
866            let expr = self.compile_expr(entity, &projection.expr, params)?;
867            let aliased = format!("{expr} AS {}", self.quote_ident(&projection.alias));
868            if !parts.contains(&aliased) {
869                parts.push(aliased);
870            }
871        }
872        for projection in query
873            .raw_projections
874            .iter()
875            .chain(query.dynamic_properties.iter())
876        {
877            let aliased = format!(
878                "{} AS {}",
879                projection.raw_sql_segment,
880                self.quote_ident(&projection.property_name)
881            );
882            if !parts.contains(&aliased) {
883                parts.push(aliased);
884            }
885        }
886        parts.extend(
887            query
888                .aggregates
889                .iter()
890                .map(|aggregate| {
891                    let field = self.resolve_aggregate_field(entity, aggregate)?;
892                    let call = self.aggregate_call_sql(aggregate.function, &field);
893                    Ok(format!("{call} AS {}", self.quote_ident(&aggregate.alias)))
894                })
895                .collect::<Result<Vec<_>, _>>()?,
896        );
897        Ok(parts.join(", "))
898    }
899
900    fn aggregate_call_sql(&self, function: AggregateFunction, field: &str) -> String {
901        let function_sql = self.aggregate_function_sql(function);
902        format!("{function_sql}({field})")
903    }
904
905    fn aggregate_function_sql(&self, function: AggregateFunction) -> &'static str {
906        match function {
907            AggregateFunction::Count => "COUNT",
908            AggregateFunction::Sum => "SUM",
909            AggregateFunction::Avg => "AVG",
910            AggregateFunction::Min => "MIN",
911            AggregateFunction::Max => "MAX",
912            AggregateFunction::Stddev => "STDDEV",
913            AggregateFunction::StddevPop => "STDDEV_POP",
914            AggregateFunction::VarSamp => "VAR_SAMP",
915            AggregateFunction::VarPop => "VAR_POP",
916            AggregateFunction::BitAnd => "BIT_AND",
917            AggregateFunction::BitOr => "BIT_OR",
918            AggregateFunction::BitXor => "BIT_XOR",
919        }
920    }
921
922    fn compile_expr(
923        &self,
924        entity: &EntityDescriptor,
925        expr: &Expr,
926        params: &mut Vec<Value>,
927    ) -> Result<String, SqlCompileError> {
928        match expr {
929            Expr::Column(name) => self.column_sql(entity, name),
930            Expr::Value(value) => {
931                params.push(value.clone());
932                Ok(self.placeholder(params.len()))
933            }
934            Expr::Function { function, args } => {
935                self.compile_function(entity, *function, args, params)
936            }
937            Expr::Binary { left, op, right } => {
938                if matches!(
939                    op,
940                    BinaryOp::In | BinaryOp::NotIn | BinaryOp::InLarge | BinaryOp::NotInLarge
941                ) {
942                    return self.compile_in(entity, left, *op, right, params);
943                }
944                let lhs = self.compile_expr(entity, left, params)?;
945                let rhs = self.compile_expr(entity, right, params)?;
946                let op = match op {
947                    BinaryOp::Eq => "=",
948                    BinaryOp::Ne => "!=",
949                    BinaryOp::Gt => ">",
950                    BinaryOp::Gte => ">=",
951                    BinaryOp::Lt => "<",
952                    BinaryOp::Lte => "<=",
953                    BinaryOp::Like => "LIKE",
954                    BinaryOp::NotLike => "NOT LIKE",
955                    BinaryOp::In | BinaryOp::NotIn | BinaryOp::InLarge | BinaryOp::NotInLarge => {
956                        unreachable!()
957                    }
958                };
959                Ok(format!("({lhs} {op} {rhs})"))
960            }
961            Expr::SubQuery {
962                left,
963                op,
964                entity: sub_entity,
965                query,
966            } => self.compile_subquery(entity, left, *op, sub_entity, query, params),
967            Expr::Between { expr, lower, upper } => {
968                let expr = self.compile_expr(entity, expr, params)?;
969                let lower = self.compile_expr(entity, lower, params)?;
970                let upper = self.compile_expr(entity, upper, params)?;
971                Ok(format!("({expr} BETWEEN {lower} AND {upper})"))
972            }
973            Expr::IsNull(expr) => {
974                let expr = self.compile_expr(entity, expr, params)?;
975                Ok(format!("({expr} IS NULL)"))
976            }
977            Expr::IsNotNull(expr) => {
978                let expr = self.compile_expr(entity, expr, params)?;
979                Ok(format!("({expr} IS NOT NULL)"))
980            }
981            Expr::And(parts) => self.compile_joined(entity, parts, "AND", params),
982            Expr::Or(parts) => self.compile_joined(entity, parts, "OR", params),
983            Expr::Not(expr) => {
984                let expr = self.compile_expr(entity, expr, params)?;
985                Ok(format!("(NOT {expr})"))
986            }
987        }
988    }
989
990    fn compile_function(
991        &self,
992        entity: &EntityDescriptor,
993        function: ExprFunction,
994        args: &[Expr],
995        params: &mut Vec<Value>,
996    ) -> Result<String, SqlCompileError> {
997        match function {
998            ExprFunction::Soundex => {
999                let [arg] = args else {
1000                    return Err(SqlCompileError::InvalidFunctionArguments(
1001                        "SOUNDEX expects exactly one argument".to_owned(),
1002                    ));
1003                };
1004                let arg = self.compile_expr(entity, arg, params)?;
1005                Ok(format!("SOUNDEX({arg})"))
1006            }
1007            ExprFunction::Gbk => self.compile_gbk_function(entity, args, params),
1008            ExprFunction::Count if args.is_empty() => Ok("COUNT(*)".to_owned()),
1009            ExprFunction::Count => self.compile_single_arg_function(entity, "COUNT", args, params),
1010            ExprFunction::Sum => self.compile_single_arg_function(entity, "SUM", args, params),
1011            ExprFunction::Avg => self.compile_single_arg_function(entity, "AVG", args, params),
1012            ExprFunction::Min => self.compile_single_arg_function(entity, "MIN", args, params),
1013            ExprFunction::Max => self.compile_single_arg_function(entity, "MAX", args, params),
1014            ExprFunction::Stddev => {
1015                self.compile_single_arg_function(entity, "STDDEV", args, params)
1016            }
1017            ExprFunction::StddevPop => {
1018                self.compile_single_arg_function(entity, "STDDEV_POP", args, params)
1019            }
1020            ExprFunction::VarSamp => {
1021                self.compile_single_arg_function(entity, "VAR_SAMP", args, params)
1022            }
1023            ExprFunction::VarPop => {
1024                self.compile_single_arg_function(entity, "VAR_POP", args, params)
1025            }
1026            ExprFunction::BitAnd => {
1027                self.compile_single_arg_function(entity, "BIT_AND", args, params)
1028            }
1029            ExprFunction::BitOr => self.compile_single_arg_function(entity, "BIT_OR", args, params),
1030            ExprFunction::BitXor => {
1031                self.compile_single_arg_function(entity, "BIT_XOR", args, params)
1032            }
1033        }
1034    }
1035
1036    fn compile_single_arg_function(
1037        &self,
1038        entity: &EntityDescriptor,
1039        function: &str,
1040        args: &[Expr],
1041        params: &mut Vec<Value>,
1042    ) -> Result<String, SqlCompileError> {
1043        let [arg] = args else {
1044            return Err(SqlCompileError::InvalidFunctionArguments(format!(
1045                "{function} expects exactly one argument"
1046            )));
1047        };
1048        let arg = self.compile_expr(entity, arg, params)?;
1049        Ok(format!("{function}({arg})"))
1050    }
1051
1052    /// Compile a GBK sort expression. The default implementation returns an error
1053    /// because GBK encoding conversion is dialect-specific. PostgreSQL dialects
1054    /// should override this to use `convert_to(arg, 'GBK')`.
1055    fn compile_gbk_function(
1056        &self,
1057        entity: &EntityDescriptor,
1058        args: &[Expr],
1059        params: &mut Vec<Value>,
1060    ) -> Result<String, SqlCompileError> {
1061        let [arg] = args else {
1062            return Err(SqlCompileError::InvalidFunctionArguments(
1063                "GBK expects exactly one argument".to_owned(),
1064            ));
1065        };
1066        // Default: pass through the column as-is (no GBK conversion).
1067        // Dialects with GBK support (e.g. PostgreSQL) should override this method.
1068        let arg = self.compile_expr(entity, arg, params)?;
1069        Ok(arg)
1070    }
1071
1072    fn compile_subquery(
1073        &self,
1074        entity: &EntityDescriptor,
1075        left: &Expr,
1076        op: BinaryOp,
1077        sub_entity: &EntityDescriptor,
1078        query: &SelectQuery,
1079        params: &mut Vec<Value>,
1080    ) -> Result<String, SqlCompileError> {
1081        let lhs = self.compile_expr(entity, left, params)?;
1082        let operator = match op {
1083            BinaryOp::In | BinaryOp::InLarge => "IN",
1084            BinaryOp::NotIn | BinaryOp::NotInLarge => "NOT IN",
1085            _ => return Err(SqlCompileError::InvalidSubQueryOperator(format!("{op:?}"))),
1086        };
1087        let subquery = self.compile_select_sql(sub_entity, query, params)?;
1088        Ok(format!("({lhs} {operator} ({subquery}))"))
1089    }
1090
1091    fn compile_joined(
1092        &self,
1093        entity: &EntityDescriptor,
1094        parts: &[Expr],
1095        joiner: &str,
1096        params: &mut Vec<Value>,
1097    ) -> Result<String, SqlCompileError> {
1098        let compiled = parts
1099            .iter()
1100            .map(|part| self.compile_expr(entity, part, params))
1101            .collect::<Result<Vec<_>, _>>()?;
1102        Ok(format!("({})", compiled.join(&format!(" {joiner} "))))
1103    }
1104
1105    fn compile_in(
1106        &self,
1107        entity: &EntityDescriptor,
1108        left: &Expr,
1109        op: BinaryOp,
1110        right: &Expr,
1111        params: &mut Vec<Value>,
1112    ) -> Result<String, SqlCompileError> {
1113        let lhs = self.compile_expr(entity, left, params)?;
1114        let operator = match op {
1115            BinaryOp::In | BinaryOp::InLarge => "IN",
1116            BinaryOp::NotIn | BinaryOp::NotInLarge => "NOT IN",
1117            _ => unreachable!(),
1118        };
1119        match right {
1120            Expr::Value(Value::List(values)) => {
1121                if values.is_empty() {
1122                    return Err(SqlCompileError::EmptyInList);
1123                }
1124                let mut placeholders = Vec::with_capacity(values.len());
1125                for value in values {
1126                    params.push(value.clone());
1127                    placeholders.push(self.placeholder(params.len()));
1128                }
1129                Ok(format!("({lhs} {operator} ({}))", placeholders.join(", ")))
1130            }
1131            _ => {
1132                let rhs = self.compile_expr(entity, right, params)?;
1133                Ok(format!("({lhs} {operator} ({rhs}))"))
1134            }
1135        }
1136    }
1137
1138    fn compile_projection(
1139        &self,
1140        entity: &EntityDescriptor,
1141        query: &SelectQuery,
1142        params: &mut Vec<Value>,
1143    ) -> Result<String, SqlCompileError> {
1144        match query.aggregates.is_empty() {
1145            true => self.select_projection(entity, query, params),
1146            false => self.aggregate_projection(entity, query, params),
1147        }
1148    }
1149
1150    fn resolve_order_field(
1151        &self,
1152        entity: &EntityDescriptor,
1153        order_by: &OrderBy,
1154        params: &mut Vec<Value>,
1155    ) -> Result<String, SqlCompileError> {
1156        match &order_by.expr {
1157            Some(expr) => self.compile_expr(entity, expr, params),
1158            None => self.column_sql(entity, &order_by.field),
1159        }
1160    }
1161
1162    fn column_with_alias(&self, property: &PropertyDescriptor) -> String {
1163        let column = self.quote_ident(&property.column_name);
1164        match property.column_name == property.name {
1165            true => column,
1166            false => format!("{column} AS {}", self.quote_ident(&property.name)),
1167        }
1168    }
1169
1170    fn resolve_aggregate_field(
1171        &self,
1172        entity: &EntityDescriptor,
1173        aggregate: &Aggregate,
1174    ) -> Result<String, SqlCompileError> {
1175        match aggregate.function == AggregateFunction::Count && aggregate.field == "*" {
1176            true => Ok("*".to_owned()),
1177            false => self.column_sql(entity, &aggregate.field),
1178        }
1179    }
1180}
1181
1182#[cfg(test)]
1183mod tests {
1184    use super::*;
1185    use crate::DatabaseKind;
1186    use teaql_core::{DataType, EntityDescriptor, InsertCommand, PropertyDescriptor, Value};
1187
1188    struct TestDialect;
1189    impl crate::SqlDialect for TestDialect {
1190        fn kind(&self) -> DatabaseKind {
1191            DatabaseKind::PostgreSql
1192        }
1193        fn quote_ident(&self, ident: &str) -> String {
1194            ident.to_owned()
1195        }
1196        fn placeholder(&self, index: usize) -> String {
1197            format!("${index}")
1198        }
1199        fn schema_type_sql(
1200            &self,
1201            _data_type: DataType,
1202            _property: &PropertyDescriptor,
1203        ) -> Result<&'static str, SqlCompileError> {
1204            Ok("TEST")
1205        }
1206    }
1207
1208    #[test]
1209    fn test_regression_issue_56_typed_null_conversion() {
1210        let dialect = TestDialect;
1211        let mut entity = EntityDescriptor::new("User");
1212        entity
1213            .properties
1214            .push(PropertyDescriptor::new("paid_at", DataType::Timestamp));
1215
1216        let mut command = InsertCommand::new("User");
1217        command = command.value("paid_at", Value::Null);
1218
1219        let query = dialect.compile_insert(&entity, &command).unwrap();
1220        // The value should be converted to TypedNull(Timestamp)
1221        assert_eq!(query.params.len(), 1);
1222        assert_eq!(query.params[0], Value::TypedNull(DataType::Timestamp));
1223    }
1224}
teaql_sql/dialect.rs

teaql_sql/
dialect.rs
