1#![allow(clippy::manual_async_fn)]
2#![allow(async_fn_in_trait)]
3
4use std::collections::HashMap;
5use std::sync::{Arc, RwLock};
6use std::time::SystemTime;
7use teaql_core::{
8 CompactRow, EntityDescriptor, EntitySnapshot, Expr, GeneratedValues, SelectQuery, Value,
9};
10use teaql_data_service::{
11 DataServiceCapabilities, DataServiceExecutor, DataServiceOperation, ExecutionMetadata,
12 MutationExecutor, MutationRequest, MutationResult, QueryExecutor, QueryRequest, QueryResult,
13};
14
15use crate::{CompiledQuery, SqlCompileError, SqlDialect};
16
17pub trait SqlTransport: Send + Sync {
18 type Error: std::error::Error + Send + Sync + 'static;
19
20 fn fetch_all_compact_sql(
21 &self,
22 query: &CompiledQuery,
23 ) -> impl std::future::Future<Output = Result<Vec<CompactRow>, Self::Error>> + Send;
24 fn fetch_repeated_compact_sql(
25 &self,
26 template: &CompiledQuery,
27 param_index: usize,
28 values: &[Value],
29 ) -> impl std::future::Future<Output = Result<Vec<CompactRow>, Self::Error>> + Send {
30 async move {
31 let mut rows = Vec::new();
32 for value in values {
33 let mut query = template.clone();
34 query.params[param_index] = value.clone();
35 rows.extend(self.fetch_all_compact_sql(&query).await?);
36 }
37 Ok(rows)
38 }
39 }
40 fn execute_sql(
41 &self,
42 query: &CompiledQuery,
43 ) -> impl std::future::Future<Output = Result<u64, Self::Error>> + Send;
44}
45
46pub trait StreamingSqlTransport: SqlTransport {
47 fn stream_sql(
48 &self,
49 query: CompiledQuery,
50 chunk_size: usize,
51 ) -> teaql_data_service::QueryStream<'_, Self::Error>;
52}
53
54pub trait SqlTransactionTransport: SqlTransport {
55 type Tx<'a>: SqlTransport<Error = Self::Error>
56 + SqlTransaction<Error = Self::Error>
57 + Send
58 + Sync
59 + 'a
60 where
61 Self: 'a;
62
63 fn begin_sql(
64 &self,
65 ) -> impl std::future::Future<Output = Result<Self::Tx<'_>, Self::Error>> + Send;
66}
67
68pub trait SqlTransaction {
69 type Error: std::error::Error + Send + Sync + 'static;
70 fn commit_sql(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
71 fn rollback_sql(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
72}
73
74#[derive(Debug)]
75pub enum SqlExecutorError<E: std::error::Error + Send + Sync + 'static> {
76 Compile(SqlCompileError),
77 Transport(E),
78 PersistedRecord(String),
79}
80
81impl<E: std::error::Error + Send + Sync + 'static> std::fmt::Display for SqlExecutorError<E> {
82 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83 match self {
84 SqlExecutorError::Compile(e) => write!(f, "SQL compile error: {}", e),
85 SqlExecutorError::Transport(e) => write!(f, "Transport error: {}", e),
86 SqlExecutorError::PersistedRecord(e) => write!(f, "Persisted record error: {}", e),
87 }
88 }
89}
90
91impl<E: std::error::Error + Send + Sync + 'static> std::error::Error for SqlExecutorError<E> {}
92
93#[derive(Clone)]
94pub struct SqlDataServiceExecutor<D, T, S> {
95 pub dialect: D,
96 pub transport: T,
97 pub schema_provider: S,
98 descriptor_cache: Arc<RwLock<HashMap<String, Arc<teaql_core::EntityDescriptor>>>>,
99 select_plan_cache: Arc<RwLock<Vec<(SelectQuery, String)>>>,
100}
101
102impl<D, T, S> SqlDataServiceExecutor<D, T, S> {
103 pub fn new(dialect: D, transport: T, schema_provider: S) -> Self {
104 Self {
105 dialect,
106 transport,
107 schema_provider,
108 descriptor_cache: Arc::new(RwLock::new(HashMap::new())),
109 select_plan_cache: Arc::new(RwLock::new(Vec::new())),
110 }
111 }
112}
113
114impl<D, T, S> SqlDataServiceExecutor<D, T, S>
115where
116 D: SqlDialect,
117 S: teaql_data_service::SchemaProvider,
118{
119 fn compile_select_cached(
120 &self,
121 entity: &EntityDescriptor,
122 query: &SelectQuery,
123 ) -> Result<CompiledQuery, SqlCompileError> {
124 compile_select_with_cache(&self.dialect, &self.select_plan_cache, entity, query)
125 }
126}
127
128fn compile_select_with_cache<D: SqlDialect>(
129 dialect: &D,
130 plan_cache: &RwLock<Vec<(SelectQuery, String)>>,
131 entity: &EntityDescriptor,
132 query: &SelectQuery,
133) -> Result<CompiledQuery, SqlCompileError> {
134 if let Ok(cache) = plan_cache.read()
135 && let Some((_, sql)) = cache
136 .iter()
137 .find(|(candidate, _)| select_plan_matches(candidate, query))
138 {
139 return Ok(CompiledQuery {
140 sql: sql.clone(),
141 params: collect_select_params(entity, query, dialect.large_in_uses_array_param()),
142 comment: query.comment.clone(),
143 });
144 }
145
146 let key = select_plan_key(query);
147 let compiled = dialect.compile_select(entity, query)?;
148 if let Ok(mut cache) = plan_cache.write() {
149 if cache.len() >= 256 {
150 cache.remove(0);
151 }
152 if !cache
153 .iter()
154 .any(|(candidate, _)| select_plan_matches(candidate, query))
155 {
156 cache.push((key, compiled.sql.clone()));
157 }
158 }
159 Ok(compiled)
160}
161
162fn select_plan_matches(key: &SelectQuery, query: &SelectQuery) -> bool {
163 key.hard_limit == query.hard_limit
164 && key.entity == query.entity
165 && key.projection == query.projection
166 && key.expr_projection.len() == query.expr_projection.len()
167 && key
168 .expr_projection
169 .iter()
170 .zip(&query.expr_projection)
171 .all(|(left, right)| {
172 left.alias == right.alias && expr_plan_matches(&left.expr, &right.expr)
173 })
174 && key.search_with_text.is_some() == query.search_with_text.is_some()
175 && optional_expr_plan_matches(key.filter.as_ref(), query.filter.as_ref())
176 && optional_expr_plan_matches(key.having.as_ref(), query.having.as_ref())
177 && key.order_by.len() == query.order_by.len()
178 && key
179 .order_by
180 .iter()
181 .zip(&query.order_by)
182 .all(|(left, right)| {
183 left.field == right.field
184 && left.direction == right.direction
185 && optional_expr_plan_matches(left.expr.as_ref(), right.expr.as_ref())
186 })
187 && key.slice == query.slice
188 && key.partition_by == query.partition_by
189 && key.aggregates == query.aggregates
190 && key.group_by == query.group_by
191 && key.relations == query.relations
192 && key.aggregation_cache == query.aggregation_cache
193 && key.raw_sql == query.raw_sql
194 && key.raw_sql_search_criteria == query.raw_sql_search_criteria
195 && key.dynamic_properties == query.dynamic_properties
196 && key.raw_projections == query.raw_projections
197 && key.object_group_bys == query.object_group_bys
198 && key.child_enhancements == query.child_enhancements
199 && key.stream_config == query.stream_config
200 && key.continuous_page_fetch == query.continuous_page_fetch
201}
202
203fn optional_expr_plan_matches(key: Option<&Expr>, query: Option<&Expr>) -> bool {
204 match (key, query) {
205 (Some(key), Some(query)) => expr_plan_matches(key, query),
206 (None, None) => true,
207 _ => false,
208 }
209}
210
211fn expr_plan_matches(key: &Expr, query: &Expr) -> bool {
212 match (key, query) {
213 (Expr::Column(left), Expr::Column(right)) => left == right,
214 (Expr::Value(Value::List(left)), Expr::Value(Value::List(right))) => {
215 left.len() == right.len()
216 }
217 (Expr::Value(Value::List(_)), Expr::Value(_))
218 | (Expr::Value(_), Expr::Value(Value::List(_))) => false,
219 (Expr::Value(_), Expr::Value(_)) => true,
220 (
221 Expr::Function {
222 function: left_function,
223 args: left_args,
224 },
225 Expr::Function {
226 function: right_function,
227 args: right_args,
228 },
229 ) => left_function == right_function && expr_slice_plan_matches(left_args, right_args),
230 (
231 Expr::Binary {
232 left: left_left,
233 op: left_op,
234 right: left_right,
235 },
236 Expr::Binary {
237 left: right_left,
238 op: right_op,
239 right: right_right,
240 },
241 ) => {
242 left_op == right_op
243 && expr_plan_matches(left_left, right_left)
244 && expr_plan_matches(left_right, right_right)
245 }
246 (
247 Expr::SubQuery {
248 left: left_expr,
249 op: left_op,
250 entity: left_entity,
251 query: left_query,
252 },
253 Expr::SubQuery {
254 left: right_expr,
255 op: right_op,
256 entity: right_entity,
257 query: right_query,
258 },
259 ) => {
260 left_op == right_op
261 && left_entity == right_entity
262 && expr_plan_matches(left_expr, right_expr)
263 && select_plan_matches(left_query, right_query)
264 }
265 (
266 Expr::Between {
267 expr: left_expr,
268 lower: left_lower,
269 upper: left_upper,
270 },
271 Expr::Between {
272 expr: right_expr,
273 lower: right_lower,
274 upper: right_upper,
275 },
276 ) => {
277 expr_plan_matches(left_expr, right_expr)
278 && expr_plan_matches(left_lower, right_lower)
279 && expr_plan_matches(left_upper, right_upper)
280 }
281 (Expr::IsNull(left), Expr::IsNull(right))
282 | (Expr::IsNotNull(left), Expr::IsNotNull(right))
283 | (Expr::Not(left), Expr::Not(right)) => expr_plan_matches(left, right),
284 (Expr::And(left), Expr::And(right)) | (Expr::Or(left), Expr::Or(right)) => {
285 expr_slice_plan_matches(left, right)
286 }
287 _ => false,
288 }
289}
290
291fn expr_slice_plan_matches(left: &[Expr], right: &[Expr]) -> bool {
292 left.len() == right.len()
293 && left
294 .iter()
295 .zip(right)
296 .all(|(left, right)| expr_plan_matches(left, right))
297}
298
299fn select_plan_key(query: &SelectQuery) -> SelectQuery {
300 let mut key = query.clone();
301 key.comment = None;
302 key.trace_chain.clear();
303 if key.search_with_text.is_some() {
304 key.search_with_text = Some(String::new());
305 }
306 for projection in &mut key.expr_projection {
307 normalize_expr_values(&mut projection.expr);
308 }
309 if let Some(expr) = &mut key.filter {
310 normalize_expr_values(expr);
311 }
312 if let Some(expr) = &mut key.having {
313 normalize_expr_values(expr);
314 }
315 for order in &mut key.order_by {
316 if let Some(expr) = &mut order.expr {
317 normalize_expr_values(expr);
318 }
319 }
320 key
321}
322
323fn normalize_expr_values(expr: &mut Expr) {
324 match expr {
325 Expr::Value(Value::List(values)) => {
326 values.fill(Value::Null);
327 }
328 Expr::Value(value) => *value = Value::Null,
329 Expr::Function { args, .. } | Expr::And(args) | Expr::Or(args) => {
330 for arg in args {
331 normalize_expr_values(arg);
332 }
333 }
334 Expr::Binary { left, right, .. } => {
335 normalize_expr_values(left);
336 normalize_expr_values(right);
337 }
338 Expr::SubQuery { left, query, .. } => {
339 normalize_expr_values(left);
340 **query = select_plan_key(query);
341 }
342 Expr::Between { expr, lower, upper } => {
343 normalize_expr_values(expr);
344 normalize_expr_values(lower);
345 normalize_expr_values(upper);
346 }
347 Expr::IsNull(expr) | Expr::IsNotNull(expr) | Expr::Not(expr) => {
348 normalize_expr_values(expr);
349 }
350 Expr::Column(_) => {}
351 }
352}
353
354fn collect_select_params(
355 entity: &EntityDescriptor,
356 query: &SelectQuery,
357 large_in_uses_array_param: bool,
358) -> Vec<Value> {
359 let mut params = Vec::new();
360 if query.raw_sql.is_some() {
361 return params;
362 }
363 for projection in &query.expr_projection {
364 collect_expr_params(&projection.expr, &mut params, large_in_uses_array_param);
365 }
366 let partitioned = query.partition_by.is_some() && query.slice.is_some();
367 if partitioned {
368 for order in &query.order_by {
369 if let Some(expr) = &order.expr {
370 collect_expr_params(expr, &mut params, large_in_uses_array_param);
371 }
372 }
373 }
374 if let Some(filter) = &query.filter {
375 collect_expr_params(filter, &mut params, large_in_uses_array_param);
376 }
377 if let Some(search_text) = &query.search_with_text {
378 let value = Value::from(format!("%{search_text}%"));
379 params.extend(
380 entity
381 .properties
382 .iter()
383 .filter(|property| {
384 matches!(
385 property.data_type,
386 teaql_core::DataType::Text | teaql_core::DataType::LargeText
387 )
388 })
389 .map(|_| value.clone()),
390 );
391 }
392 if partitioned {
393 return params;
394 }
395 if let Some(having) = &query.having {
396 collect_expr_params(having, &mut params, large_in_uses_array_param);
397 }
398 for order in &query.order_by {
399 if let Some(expr) = &order.expr {
400 collect_expr_params(expr, &mut params, large_in_uses_array_param);
401 }
402 }
403 params
404}
405
406fn collect_expr_params(expr: &Expr, params: &mut Vec<Value>, large_in_uses_array_param: bool) {
407 match expr {
408 Expr::Column(_) => {}
409 Expr::Value(value) => params.push(value.clone()),
410 Expr::Function { args, .. } | Expr::And(args) | Expr::Or(args) => {
411 for arg in args {
412 collect_expr_params(arg, params, large_in_uses_array_param);
413 }
414 }
415 Expr::Binary { left, op, right } => {
416 collect_expr_params(left, params, large_in_uses_array_param);
417 if let Expr::Value(Value::List(values)) = right.as_ref()
418 && matches!(
419 op,
420 teaql_core::BinaryOp::In
421 | teaql_core::BinaryOp::NotIn
422 | teaql_core::BinaryOp::InLarge
423 | teaql_core::BinaryOp::NotInLarge
424 )
425 {
426 if large_in_uses_array_param
427 && matches!(
428 op,
429 teaql_core::BinaryOp::InLarge | teaql_core::BinaryOp::NotInLarge
430 )
431 {
432 params.push(Value::List(values.clone()));
433 } else {
434 params.extend(values.iter().cloned());
435 }
436 } else {
437 collect_expr_params(right, params, large_in_uses_array_param);
438 }
439 }
440 Expr::SubQuery {
441 left,
442 entity,
443 query,
444 ..
445 } => {
446 collect_expr_params(left, params, large_in_uses_array_param);
447 params.extend(collect_select_params(
448 entity,
449 query,
450 large_in_uses_array_param,
451 ));
452 }
453 Expr::Between { expr, lower, upper } => {
454 collect_expr_params(expr, params, large_in_uses_array_param);
455 collect_expr_params(lower, params, large_in_uses_array_param);
456 collect_expr_params(upper, params, large_in_uses_array_param);
457 }
458 Expr::IsNull(expr) | Expr::IsNotNull(expr) | Expr::Not(expr) => {
459 collect_expr_params(expr, params, large_in_uses_array_param);
460 }
461 }
462}
463
464fn partition_probe_values(query: &SelectQuery) -> Option<Vec<Value>> {
465 let field = query.partition_by.as_deref()?;
466 fn find(expr: &Expr, field: &str) -> Option<Vec<Value>> {
467 match expr {
468 Expr::Binary { left, op, right }
469 if matches!(op, teaql_core::BinaryOp::In | teaql_core::BinaryOp::InLarge)
470 && matches!(left.as_ref(), Expr::Column(column) if column == field) =>
471 {
472 match right.as_ref() {
473 Expr::Value(Value::List(values)) => Some(values.clone()),
474 _ => None,
475 }
476 }
477 Expr::And(parts) => parts.iter().find_map(|part| find(part, field)),
478 _ => None,
479 }
480 }
481 find(query.filter.as_ref()?, field)
482}
483
484fn scalar_partition_probe_query(query: &SelectQuery, value: Value) -> Option<SelectQuery> {
485 let field = query.partition_by.as_deref()?;
486 fn replace(expr: &mut Expr, field: &str, value: &Value) -> bool {
487 match expr {
488 Expr::Binary { left, op, right }
489 if matches!(op, teaql_core::BinaryOp::In | teaql_core::BinaryOp::InLarge)
490 && matches!(left.as_ref(), Expr::Column(column) if column == field) =>
491 {
492 *op = teaql_core::BinaryOp::Eq;
493 **right = Expr::Value(value.clone());
494 true
495 }
496 Expr::And(parts) => parts.iter_mut().any(|part| replace(part, field, value)),
497 _ => false,
498 }
499 }
500
501 let mut scalar = query.clone();
502 if !replace(scalar.filter.as_mut()?, field, &value) {
503 return None;
504 }
505 scalar.partition_by = None;
506 Some(scalar)
507}
508
509impl<D, T, S> SqlDataServiceExecutor<D, T, S>
510where
511 S: teaql_data_service::SchemaProvider,
512{
513 fn entity_descriptor(&self, name: &str) -> Option<Arc<teaql_core::EntityDescriptor>> {
514 if let Ok(cache) = self.descriptor_cache.read()
515 && let Some(descriptor) = cache.get(name)
516 {
517 return Some(descriptor.clone());
518 }
519 let descriptor = self.schema_provider.get_entity(name)?;
520 if let Ok(mut cache) = self.descriptor_cache.write() {
521 return Some(
522 cache
523 .entry(name.to_owned())
524 .or_insert_with(|| descriptor.clone())
525 .clone(),
526 );
527 }
528 Some(descriptor)
529 }
530}
531
532impl<
533 D: SqlDialect + Send + Sync,
534 T: SqlTransport + Send + Sync,
535 S: teaql_data_service::SchemaProvider + Send + Sync,
536> DataServiceExecutor for SqlDataServiceExecutor<D, T, S>
537{
538 type Error = SqlExecutorError<T::Error>;
539
540 fn capabilities(&self) -> DataServiceCapabilities {
541 DataServiceCapabilities {
542 query: true,
543 mutation: true,
544 transaction: false, schema: false,
546 id_generation: false,
547 batch_mutation: true,
548 returning: false,
549 small_parent_relation_probes: self.dialect.prefers_small_parent_relation_probes(),
550 }
551 }
552}
553
554#[cfg(test)]
555mod tests {
556 use super::*;
557 use std::sync::atomic::{AtomicUsize, Ordering};
558 use teaql_core::{DataType, EntityDescriptor, PropertyDescriptor};
559
560 #[derive(Clone, Copy)]
561 struct TestDialect;
562
563 impl SqlDialect for TestDialect {
564 fn kind(&self) -> crate::DatabaseKind {
565 crate::DatabaseKind::PostgreSql
566 }
567
568 fn quote_ident(&self, ident: &str) -> String {
569 format!("\"{ident}\"")
570 }
571
572 fn placeholder(&self, index: usize) -> String {
573 format!("${index}")
574 }
575 }
576
577 #[derive(Clone, Copy)]
578 struct ArrayTestDialect;
579
580 impl SqlDialect for ArrayTestDialect {
581 fn kind(&self) -> crate::DatabaseKind {
582 crate::DatabaseKind::PostgreSql
583 }
584
585 fn quote_ident(&self, ident: &str) -> String {
586 format!("\"{ident}\"")
587 }
588
589 fn placeholder(&self, index: usize) -> String {
590 format!("${index}")
591 }
592
593 fn large_in_uses_array_param(&self) -> bool {
594 true
595 }
596
597 fn compile_in(
598 &self,
599 entity: &EntityDescriptor,
600 left: &Expr,
601 op: teaql_core::BinaryOp,
602 right: &Expr,
603 params: &mut Vec<Value>,
604 ) -> Result<String, SqlCompileError> {
605 if matches!(
606 op,
607 teaql_core::BinaryOp::InLarge | teaql_core::BinaryOp::NotInLarge
608 ) && let Expr::Value(Value::List(values)) = right
609 {
610 let lhs = self.compile_expr(entity, left, params)?;
611 params.push(Value::List(values.clone()));
612 let operator = if op == teaql_core::BinaryOp::InLarge {
613 "= ANY"
614 } else {
615 "<> ALL"
616 };
617 return Ok(format!("({lhs} {operator}(${}))", params.len()));
618 }
619 Err(SqlCompileError::InvalidFunctionArguments(
620 "array test dialect only supports large IN".to_owned(),
621 ))
622 }
623 }
624
625 #[derive(Clone, Copy)]
626 struct EmptyTransport;
627
628 impl SqlTransport for EmptyTransport {
629 type Error = std::io::Error;
630
631 async fn fetch_all_compact_sql(
632 &self,
633 _query: &CompiledQuery,
634 ) -> Result<Vec<CompactRow>, Self::Error> {
635 Ok(Vec::new())
636 }
637
638 async fn execute_sql(&self, _query: &CompiledQuery) -> Result<u64, Self::Error> {
639 Ok(0)
640 }
641 }
642
643 #[derive(Clone)]
644 struct RepeatedProbeTransport {
645 calls: Arc<AtomicUsize>,
646 single_calls: Arc<AtomicUsize>,
647 }
648
649 impl SqlTransport for RepeatedProbeTransport {
650 type Error = std::io::Error;
651
652 async fn fetch_all_compact_sql(
653 &self,
654 _query: &CompiledQuery,
655 ) -> Result<Vec<CompactRow>, Self::Error> {
656 self.single_calls.fetch_add(1, Ordering::Relaxed);
657 Ok(Vec::new())
658 }
659
660 async fn fetch_repeated_compact_sql(
661 &self,
662 template: &CompiledQuery,
663 param_index: usize,
664 values: &[Value],
665 ) -> Result<Vec<CompactRow>, Self::Error> {
666 self.calls.fetch_add(1, Ordering::Relaxed);
667 assert_eq!(values, [Value::U64(7), Value::U64(9)]);
668 assert_eq!(template.params[param_index], Value::U64(7));
669 Ok(Vec::new())
670 }
671
672 async fn execute_sql(&self, _query: &CompiledQuery) -> Result<u64, Self::Error> {
673 Ok(0)
674 }
675 }
676
677 #[derive(Clone, Copy)]
678 struct ProbeDialect;
679
680 impl SqlDialect for ProbeDialect {
681 fn kind(&self) -> crate::DatabaseKind {
682 crate::DatabaseKind::Sqlite
683 }
684
685 fn quote_ident(&self, ident: &str) -> String {
686 format!("\"{ident}\"")
687 }
688
689 fn placeholder(&self, _index: usize) -> String {
690 "?".to_owned()
691 }
692
693 fn prefers_small_parent_relation_probes(&self) -> bool {
694 true
695 }
696 }
697
698 #[derive(Clone)]
699 struct CountingSchemaProvider {
700 lookups: Arc<AtomicUsize>,
701 }
702
703 impl teaql_data_service::SchemaProvider for CountingSchemaProvider {
704 fn get_entity(&self, name: &str) -> Option<Arc<EntityDescriptor>> {
705 self.lookups.fetch_add(1, Ordering::Relaxed);
706 (name == "Order").then(|| Arc::new(test_entity()))
707 }
708 }
709
710 fn test_entity() -> EntityDescriptor {
711 EntityDescriptor::new("Order")
712 .property(PropertyDescriptor::new("id", DataType::U64).id().not_null())
713 .property(PropertyDescriptor::new("name", DataType::Text))
714 }
715
716 fn query_request(capture_debug_query: bool) -> QueryRequest {
717 QueryRequest {
718 query: SelectQuery::new("Order"),
719 trace_chain: Vec::new(),
720 comment: None,
721 capture_debug_query,
722 capture_execution_metadata: true,
723 }
724 }
725
726 #[tokio::test]
727 async fn caches_entity_descriptors_across_executor_clones() {
728 let lookups = Arc::new(AtomicUsize::new(0));
729 let executor = SqlDataServiceExecutor::new(
730 TestDialect,
731 EmptyTransport,
732 CountingSchemaProvider {
733 lookups: lookups.clone(),
734 },
735 );
736
737 let result = executor.query(query_request(false)).await.unwrap();
738 executor.clone().query(query_request(true)).await.unwrap();
739
740 assert_eq!(lookups.load(Ordering::Relaxed), 1);
741 assert!(result.metadata.debug_query.is_none());
742 }
743
744 #[tokio::test]
745 async fn skips_execution_metadata_when_caller_will_discard_it() {
746 let executor = SqlDataServiceExecutor::new(
747 TestDialect,
748 EmptyTransport,
749 CountingSchemaProvider {
750 lookups: Arc::new(AtomicUsize::new(0)),
751 },
752 );
753 let mut request = query_request(false);
754 request.capture_execution_metadata = false;
755
756 let result = executor.query(request).await.unwrap();
757
758 assert!(result.metadata.backend.is_empty());
759 assert_eq!(result.metadata.started_at, SystemTime::UNIX_EPOCH);
760 assert!(result.metadata.parameterized_query.is_none());
761 assert!(result.metadata.params.is_empty());
762 assert!(result.metadata.trace_chain.is_empty());
763 }
764
765 #[tokio::test]
766 async fn executes_trace_off_partition_query_as_one_repeated_probe() {
767 let calls = Arc::new(AtomicUsize::new(0));
768 let single_calls = Arc::new(AtomicUsize::new(0));
769 let executor = SqlDataServiceExecutor::new(
770 ProbeDialect,
771 RepeatedProbeTransport {
772 calls: calls.clone(),
773 single_calls: single_calls.clone(),
774 },
775 CountingSchemaProvider {
776 lookups: Arc::new(AtomicUsize::new(0)),
777 },
778 );
779 let mut request = query_request(false);
780 request.capture_execution_metadata = false;
781 request.query = request
782 .query
783 .filter(Expr::in_list("id", [Value::U64(7), Value::U64(9)]))
784 .order_desc("id")
785 .limit(1)
786 .partition_by("id");
787
788 let result = executor.query(request).await.unwrap();
789
790 assert!(result.rows.is_empty());
791 assert_eq!(calls.load(Ordering::Relaxed), 1);
792 assert_eq!(single_calls.load(Ordering::Relaxed), 0);
793 assert_eq!(result.metadata.started_at, SystemTime::UNIX_EPOCH);
794 }
795
796 #[tokio::test]
797 async fn keeps_partition_query_observable_when_metadata_is_enabled() {
798 let calls = Arc::new(AtomicUsize::new(0));
799 let single_calls = Arc::new(AtomicUsize::new(0));
800 let executor = SqlDataServiceExecutor::new(
801 ProbeDialect,
802 RepeatedProbeTransport {
803 calls: calls.clone(),
804 single_calls: single_calls.clone(),
805 },
806 CountingSchemaProvider {
807 lookups: Arc::new(AtomicUsize::new(0)),
808 },
809 );
810 let mut request = query_request(false);
811 request.query = request
812 .query
813 .filter(Expr::in_list("id", [Value::U64(7), Value::U64(9)]))
814 .order_desc("id")
815 .limit(1)
816 .partition_by("id");
817
818 executor.query(request).await.unwrap();
819
820 assert_eq!(calls.load(Ordering::Relaxed), 0);
821 assert_eq!(single_calls.load(Ordering::Relaxed), 1);
822 }
823
824 #[tokio::test]
825 async fn cached_select_plan_rebinds_values_and_separates_in_list_lengths() {
826 let lookups = Arc::new(AtomicUsize::new(0));
827 let executor = SqlDataServiceExecutor::new(
828 TestDialect,
829 EmptyTransport,
830 CountingSchemaProvider { lookups },
831 );
832 let request = |filter| QueryRequest {
833 query: SelectQuery::new("Order").filter(filter),
834 trace_chain: Vec::new(),
835 comment: None,
836 capture_debug_query: false,
837 capture_execution_metadata: true,
838 };
839
840 let first = executor
841 .query(request(Expr::eq("id", 7_u64)))
842 .await
843 .unwrap();
844 let second = executor
845 .query(request(Expr::eq("id", 9_u64)))
846 .await
847 .unwrap();
848 assert_eq!(
849 first.metadata.parameterized_query,
850 second.metadata.parameterized_query
851 );
852 assert_eq!(first.metadata.params, vec![Value::U64(7)]);
853 assert_eq!(second.metadata.params, vec![Value::U64(9)]);
854
855 let short = executor
856 .query(request(Expr::in_list("id", [Value::U64(1), Value::U64(2)])))
857 .await
858 .unwrap();
859 let long = executor
860 .query(request(Expr::in_list(
861 "id",
862 [Value::U64(1), Value::U64(2), Value::U64(3)],
863 )))
864 .await
865 .unwrap();
866 assert_ne!(
867 short.metadata.parameterized_query,
868 long.metadata.parameterized_query
869 );
870 assert_eq!(short.metadata.params.len(), 2);
871 assert_eq!(long.metadata.params.len(), 3);
872 }
873
874 #[tokio::test]
875 async fn cached_select_plan_rebinds_large_in_as_one_array_parameter() {
876 let executor = SqlDataServiceExecutor::new(
877 ArrayTestDialect,
878 EmptyTransport,
879 CountingSchemaProvider {
880 lookups: Arc::new(AtomicUsize::new(0)),
881 },
882 );
883 let request = |values: Vec<Value>| QueryRequest {
884 query: SelectQuery::new("Order").filter(Expr::in_list("id", values)),
885 trace_chain: Vec::new(),
886 comment: None,
887 capture_debug_query: false,
888 capture_execution_metadata: true,
889 };
890 let first_values = (1_u64..=21).map(Value::from).collect::<Vec<_>>();
891 let second_values = (101_u64..=121).map(Value::from).collect::<Vec<_>>();
892
893 let first = executor.query(request(first_values.clone())).await.unwrap();
894 let second = executor
895 .query(request(second_values.clone()))
896 .await
897 .unwrap();
898
899 assert_eq!(
900 first.metadata.parameterized_query,
901 second.metadata.parameterized_query
902 );
903 assert_eq!(first.metadata.params, vec![Value::List(first_values)]);
904 assert_eq!(second.metadata.params, vec![Value::List(second_values)]);
905 }
906
907 #[tokio::test]
908 async fn cached_select_plan_preserves_parameter_order_for_supported_query_shapes() {
909 let executor = SqlDataServiceExecutor::new(
910 TestDialect,
911 EmptyTransport,
912 CountingSchemaProvider {
913 lookups: Arc::new(AtomicUsize::new(0)),
914 },
915 );
916
917 async fn assert_rebound(
918 executor: &SqlDataServiceExecutor<TestDialect, EmptyTransport, CountingSchemaProvider>,
919 warm: SelectQuery,
920 current: SelectQuery,
921 ) {
922 let request = |query| QueryRequest {
923 query,
924 trace_chain: Vec::new(),
925 comment: None,
926 capture_debug_query: false,
927 capture_execution_metadata: true,
928 };
929 executor.query(request(warm)).await.unwrap();
930 let actual = executor.query(request(current.clone())).await.unwrap();
931 let expected = TestDialect
932 .compile_select(&test_entity(), ¤t)
933 .unwrap();
934 assert_eq!(actual.metadata.parameterized_query, Some(expected.sql));
935 assert_eq!(actual.metadata.params, expected.params);
936 }
937
938 assert_rebound(
939 &executor,
940 SelectQuery::new("Order").search_with_text("first"),
941 SelectQuery::new("Order").search_with_text("second"),
942 )
943 .await;
944 assert_rebound(
945 &executor,
946 SelectQuery::new("Order")
947 .project_expr("marker", Expr::value(1_i64))
948 .filter(Expr::eq("id", 2_u64))
949 .having(Expr::gt("id", 3_u64))
950 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(4_i64))),
951 SelectQuery::new("Order")
952 .project_expr("marker", Expr::value(11_i64))
953 .filter(Expr::eq("id", 12_u64))
954 .having(Expr::gt("id", 13_u64))
955 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(14_i64))),
956 )
957 .await;
958 assert_rebound(
959 &executor,
960 SelectQuery::new("Order")
961 .filter(Expr::eq("id", 1_u64))
962 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(2_i64)))
963 .page(0, 10)
964 .partition_by("name"),
965 SelectQuery::new("Order")
966 .filter(Expr::eq("id", 3_u64))
967 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(4_i64)))
968 .page(0, 10)
969 .partition_by("name"),
970 )
971 .await;
972 assert_rebound(
973 &executor,
974 SelectQuery::new("Order").filter(Expr::in_subquery(
975 "id",
976 test_entity(),
977 SelectQuery::new("Order").filter(Expr::gt("id", 20_u64)),
978 "id",
979 )),
980 SelectQuery::new("Order").filter(Expr::in_subquery(
981 "id",
982 test_entity(),
983 SelectQuery::new("Order").filter(Expr::gt("id", 30_u64)),
984 "id",
985 )),
986 )
987 .await;
988 }
989}
990
991impl<
992 D: SqlDialect + Send + Sync,
993 T: SqlTransport + Send + Sync,
994 S: teaql_data_service::SchemaProvider + Send + Sync,
995> QueryExecutor for SqlDataServiceExecutor<D, T, S>
996{
997 fn query(
998 &self,
999 request: QueryRequest,
1000 ) -> impl std::future::Future<Output = Result<QueryResult, Self::Error>> + Send {
1001 async move {
1002 let entity_desc = self
1003 .entity_descriptor(&request.query.entity)
1004 .ok_or_else(|| {
1005 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1006 request.query.entity.clone(),
1007 ))
1008 })?;
1009
1010 if !request.capture_execution_metadata
1011 && self.dialect.prefers_small_parent_relation_probes()
1012 && let Some(values) = partition_probe_values(&request.query)
1013 && values.len() >= 2
1014 && let (Some(first_query), Some(second_query)) = (
1015 scalar_partition_probe_query(&request.query, values[0].clone()),
1016 scalar_partition_probe_query(&request.query, values[1].clone()),
1017 )
1018 {
1019 let first = self
1020 .compile_select_cached(&entity_desc, &first_query)
1021 .map_err(SqlExecutorError::Compile)?;
1022 let second_params = collect_select_params(
1023 &entity_desc,
1024 &second_query,
1025 self.dialect.large_in_uses_array_param(),
1026 );
1027 if let Some(param_index) = first
1028 .params
1029 .iter()
1030 .zip(&second_params)
1031 .position(|(left, right)| left != right)
1032 {
1033 let rows = self
1034 .transport
1035 .fetch_repeated_compact_sql(&first, param_index, &values)
1036 .await
1037 .map_err(SqlExecutorError::Transport)?;
1038 return Ok(QueryResult {
1039 metadata: ExecutionMetadata::unrecorded_query(rows.len()),
1040 rows,
1041 });
1042 }
1043 }
1044
1045 let compiled = self
1046 .compile_select_cached(&entity_desc, &request.query)
1047 .map_err(SqlExecutorError::Compile)?;
1048 let start = request.capture_execution_metadata.then(SystemTime::now);
1049 let rows = self
1050 .transport
1051 .fetch_all_compact_sql(&compiled)
1052 .await
1053 .map_err(SqlExecutorError::Transport)?;
1054 let end = request.capture_execution_metadata.then(SystemTime::now);
1055 let debug_query = request
1056 .capture_debug_query
1057 .then(|| compiled.debug_sql(self.dialect.kind()));
1058 let metadata = if request.capture_execution_metadata {
1059 let CompiledQuery { sql, params, .. } = compiled;
1060 ExecutionMetadata {
1061 backend: "sql".to_string(),
1062 operation: DataServiceOperation::Query,
1063 started_at: start.expect("captured query start"),
1064 ended_at: end.expect("captured query end"),
1065 affected_rows: None,
1066 result_count: Some(rows.len()),
1067 trace_chain: request.trace_chain,
1068 comment: request.comment,
1069 backend_request_id: None,
1070 parameterized_query: Some(sql),
1071 params,
1072 debug_query,
1073 }
1074 } else {
1075 ExecutionMetadata::unrecorded_query(rows.len())
1076 };
1077
1078 Ok(QueryResult { rows, metadata })
1079 }
1080 }
1081}
1082
1083impl<
1084 D: SqlDialect + Send + Sync,
1085 T: SqlTransport + Send + Sync,
1086 S: teaql_data_service::SchemaProvider + Send + Sync,
1087> MutationExecutor for SqlDataServiceExecutor<D, T, S>
1088{
1089 fn mutate(
1090 &self,
1091 request: MutationRequest,
1092 ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send {
1093 async move {
1094 let entity_name = match &request {
1095 MutationRequest::Insert(cmd) => &cmd.entity,
1096 MutationRequest::Update(cmd) => &cmd.entity,
1097 MutationRequest::Delete(cmd) => &cmd.entity,
1098 MutationRequest::Recover(cmd) => &cmd.entity,
1099 MutationRequest::Batch(mutations) => {
1100 let mut total_affected = 0;
1101 let mut parameterized_queries = Vec::new();
1102 let mut params = Vec::new();
1103 let mut debug_queries = Vec::new();
1104 let start = SystemTime::now();
1105 for req in mutations {
1106 let res = Box::pin(self.mutate(req.clone())).await?;
1107 total_affected += res.affected_rows;
1108 if let Some(query) = res.metadata.parameterized_query {
1109 parameterized_queries.push(query);
1110 }
1111 params.extend(res.metadata.params);
1112 if let Some(query) = res.metadata.debug_query {
1113 debug_queries.push(query);
1114 }
1115 }
1116 let end = SystemTime::now();
1117 return Ok(MutationResult {
1118 affected_rows: total_affected,
1119 generated_values: GeneratedValues::default(),
1120 persisted_snapshot: None,
1121 metadata: ExecutionMetadata {
1122 backend: "sql".to_string(),
1123 operation: DataServiceOperation::Batch,
1124 started_at: start,
1125 ended_at: end,
1126 affected_rows: Some(total_affected),
1127 result_count: None,
1128 trace_chain: Vec::new(),
1129 comment: None,
1130 backend_request_id: None,
1131 parameterized_query: (!parameterized_queries.is_empty())
1132 .then(|| parameterized_queries.join("; ")),
1133 params,
1134 debug_query: (!debug_queries.is_empty())
1135 .then(|| debug_queries.join("; ")),
1136 },
1137 });
1138 }
1139 };
1140
1141 let entity_desc = self.entity_descriptor(entity_name).ok_or_else(|| {
1142 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(entity_name.clone()))
1143 })?;
1144
1145 let compiled = match &request {
1146 MutationRequest::Insert(cmd) => self
1147 .dialect
1148 .compile_insert(&entity_desc, cmd)
1149 .map_err(SqlExecutorError::Compile)?,
1150 MutationRequest::Update(cmd) => self
1151 .dialect
1152 .compile_update(&entity_desc, cmd)
1153 .map_err(SqlExecutorError::Compile)?,
1154 MutationRequest::Delete(cmd) => self
1155 .dialect
1156 .compile_delete(&entity_desc, cmd)
1157 .map_err(SqlExecutorError::Compile)?,
1158 MutationRequest::Recover(cmd) => self
1159 .dialect
1160 .compile_recover(&entity_desc, cmd)
1161 .map_err(SqlExecutorError::Compile)?,
1162 MutationRequest::Batch(_) => unreachable!(),
1163 };
1164
1165 let start = SystemTime::now();
1166 let affected_rows = self
1167 .transport
1168 .execute_sql(&compiled)
1169 .await
1170 .map_err(SqlExecutorError::Transport)?;
1171 let end = SystemTime::now();
1172
1173 let operation = match &request {
1174 MutationRequest::Insert(_) => DataServiceOperation::Insert,
1175 MutationRequest::Update(_) => DataServiceOperation::Update,
1176 MutationRequest::Delete(_) => DataServiceOperation::Delete,
1177 MutationRequest::Recover(_) => DataServiceOperation::Recover,
1178 MutationRequest::Batch(_) => DataServiceOperation::Batch,
1179 };
1180
1181 let metadata = ExecutionMetadata {
1182 backend: "sql".to_string(),
1183 operation,
1184 started_at: start,
1185 ended_at: end,
1186 affected_rows: Some(affected_rows),
1187 result_count: None,
1188 trace_chain: request.trace_chain().to_vec(),
1189 comment: request.comment().map(|s| s.to_owned()),
1190 backend_request_id: None,
1191 parameterized_query: Some(compiled.sql.clone()),
1192 params: compiled.params.clone(),
1193 debug_query: Some(compiled.debug_sql(self.dialect.kind())),
1194 };
1195
1196 Ok(MutationResult {
1197 affected_rows,
1198 generated_values: GeneratedValues::default(),
1199 persisted_snapshot: None,
1200 metadata,
1201 })
1202 }
1203 }
1204}
1205
1206#[derive(Clone)]
1207pub struct SqlDataServiceTransaction<'a, D, Tx: SqlTransport + SqlTransaction, S> {
1208 pub dialect: &'a D,
1209 pub transport: Tx,
1210 pub schema_provider: &'a S,
1211 descriptor_cache: Arc<RwLock<HashMap<String, Arc<teaql_core::EntityDescriptor>>>>,
1212 select_plan_cache: Arc<RwLock<Vec<(SelectQuery, String)>>>,
1213}
1214
1215impl<'a, D, Tx: SqlTransport + SqlTransaction, S> SqlDataServiceTransaction<'a, D, Tx, S>
1216where
1217 S: teaql_data_service::SchemaProvider,
1218{
1219 fn entity_descriptor(&self, name: &str) -> Option<Arc<teaql_core::EntityDescriptor>> {
1220 if let Ok(cache) = self.descriptor_cache.read()
1221 && let Some(descriptor) = cache.get(name)
1222 {
1223 return Some(descriptor.clone());
1224 }
1225 let descriptor = self.schema_provider.get_entity(name)?;
1226 if let Ok(mut cache) = self.descriptor_cache.write() {
1227 return Some(
1228 cache
1229 .entry(name.to_owned())
1230 .or_insert_with(|| descriptor.clone())
1231 .clone(),
1232 );
1233 }
1234 Some(descriptor)
1235 }
1236
1237 fn compile_select_cached(
1238 &self,
1239 entity: &EntityDescriptor,
1240 query: &SelectQuery,
1241 ) -> Result<CompiledQuery, SqlCompileError>
1242 where
1243 D: SqlDialect,
1244 {
1245 compile_select_with_cache(self.dialect, &self.select_plan_cache, entity, query)
1246 }
1247}
1248
1249impl<
1250 'a,
1251 D: SqlDialect + Send + Sync,
1252 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1253 S: teaql_data_service::SchemaProvider + Send + Sync,
1254> DataServiceExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1255{
1256 type Error = SqlExecutorError<<Tx as SqlTransport>::Error>;
1257
1258 fn capabilities(&self) -> DataServiceCapabilities {
1259 DataServiceCapabilities {
1260 query: true,
1261 mutation: true,
1262 transaction: false,
1263 schema: false,
1264 id_generation: false,
1265 batch_mutation: true,
1266 returning: false,
1267 small_parent_relation_probes: self.dialect.prefers_small_parent_relation_probes(),
1268 }
1269 }
1270}
1271
1272impl<
1273 'a,
1274 D: SqlDialect + Send + Sync,
1275 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1276 S: teaql_data_service::SchemaProvider + Send + Sync,
1277> QueryExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1278{
1279 fn query(
1280 &self,
1281 request: QueryRequest,
1282 ) -> impl std::future::Future<Output = Result<QueryResult, Self::Error>> + Send {
1283 async move {
1284 let entity_desc = self
1285 .entity_descriptor(&request.query.entity)
1286 .ok_or_else(|| {
1287 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1288 request.query.entity.clone(),
1289 ))
1290 })?;
1291
1292 let compiled = self
1293 .compile_select_cached(&entity_desc, &request.query)
1294 .map_err(SqlExecutorError::Compile)?;
1295 let start = SystemTime::now();
1296 let rows = self
1297 .transport
1298 .fetch_all_compact_sql(&compiled)
1299 .await
1300 .map_err(SqlExecutorError::Transport)?;
1301 let end = SystemTime::now();
1302
1303 let metadata = ExecutionMetadata {
1304 backend: "sql".to_string(),
1305 operation: DataServiceOperation::Query,
1306 started_at: start,
1307 ended_at: end,
1308 affected_rows: None,
1309 result_count: Some(rows.len()),
1310 trace_chain: request.trace_chain,
1311 comment: request.comment,
1312 backend_request_id: None,
1313 parameterized_query: Some(compiled.sql.clone()),
1314 params: compiled.params.clone(),
1315 debug_query: request
1316 .capture_debug_query
1317 .then(|| compiled.debug_sql(self.dialect.kind())),
1318 };
1319
1320 Ok(QueryResult { rows, metadata })
1321 }
1322 }
1323}
1324
1325impl<
1326 'a,
1327 D: SqlDialect + Send + Sync,
1328 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1329 S: teaql_data_service::SchemaProvider + Send + Sync,
1330> MutationExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1331{
1332 fn mutate(
1333 &self,
1334 request: MutationRequest,
1335 ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send {
1336 async move {
1337 let entity_name = match &request {
1338 MutationRequest::Insert(cmd) => &cmd.entity,
1339 MutationRequest::Update(cmd) => &cmd.entity,
1340 MutationRequest::Delete(cmd) => &cmd.entity,
1341 MutationRequest::Recover(cmd) => &cmd.entity,
1342 MutationRequest::Batch(mutations) => {
1343 let mut total_affected = 0;
1344 let mut parameterized_queries = Vec::new();
1345 let mut params = Vec::new();
1346 let mut debug_queries = Vec::new();
1347 let start = SystemTime::now();
1348 for req in mutations {
1349 let res = Box::pin(self.mutate(req.clone())).await?;
1350 total_affected += res.affected_rows;
1351 if let Some(query) = res.metadata.parameterized_query {
1352 parameterized_queries.push(query);
1353 }
1354 params.extend(res.metadata.params);
1355 if let Some(query) = res.metadata.debug_query {
1356 debug_queries.push(query);
1357 }
1358 }
1359 let end = SystemTime::now();
1360 return Ok(MutationResult {
1361 affected_rows: total_affected,
1362 generated_values: GeneratedValues::default(),
1363 persisted_snapshot: None,
1364 metadata: ExecutionMetadata {
1365 backend: "sql".to_string(),
1366 operation: DataServiceOperation::Batch,
1367 started_at: start,
1368 ended_at: end,
1369 affected_rows: Some(total_affected),
1370 result_count: None,
1371 trace_chain: Vec::new(),
1372 comment: None,
1373 backend_request_id: None,
1374 parameterized_query: (!parameterized_queries.is_empty())
1375 .then(|| parameterized_queries.join("; ")),
1376 params,
1377 debug_query: (!debug_queries.is_empty())
1378 .then(|| debug_queries.join("; ")),
1379 },
1380 });
1381 }
1382 };
1383
1384 let entity_desc = self.entity_descriptor(entity_name).ok_or_else(|| {
1385 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(entity_name.clone()))
1386 })?;
1387
1388 let compiled = match &request {
1389 MutationRequest::Insert(cmd) => self
1390 .dialect
1391 .compile_insert(&entity_desc, cmd)
1392 .map_err(SqlExecutorError::Compile)?,
1393 MutationRequest::Update(cmd) => self
1394 .dialect
1395 .compile_update(&entity_desc, cmd)
1396 .map_err(SqlExecutorError::Compile)?,
1397 MutationRequest::Delete(cmd) => self
1398 .dialect
1399 .compile_delete(&entity_desc, cmd)
1400 .map_err(SqlExecutorError::Compile)?,
1401 MutationRequest::Recover(cmd) => self
1402 .dialect
1403 .compile_recover(&entity_desc, cmd)
1404 .map_err(SqlExecutorError::Compile)?,
1405 MutationRequest::Batch(_) => unreachable!("batch handled above"),
1406 };
1407
1408 let start = SystemTime::now();
1409 let affected_rows = self
1410 .transport
1411 .execute_sql(&compiled)
1412 .await
1413 .map_err(SqlExecutorError::Transport)?;
1414 let end = SystemTime::now();
1415
1416 let operation = match &request {
1417 MutationRequest::Insert(_) => DataServiceOperation::Insert,
1418 MutationRequest::Update(_) => DataServiceOperation::Update,
1419 MutationRequest::Delete(_) => DataServiceOperation::Delete,
1420 MutationRequest::Recover(_) => DataServiceOperation::Recover,
1421 MutationRequest::Batch(_) => DataServiceOperation::Batch,
1422 };
1423
1424 let persisted_id = match &request {
1425 MutationRequest::Insert(cmd) => cmd.values.get("id").cloned(),
1426 MutationRequest::Update(cmd) => Some(cmd.id.clone()),
1427 MutationRequest::Delete(cmd) if cmd.soft_delete => Some(cmd.id.clone()),
1428 MutationRequest::Recover(cmd) => Some(cmd.id.clone()),
1429 MutationRequest::Delete(_) | MutationRequest::Batch(_) => None,
1430 };
1431 let persisted_snapshot = if affected_rows == 1 {
1432 if let Some(id) = persisted_id {
1433 let query = SelectQuery::new(entity_name.clone()).filter(Expr::eq("id", id));
1434 let compiled_readback = self
1435 .compile_select_cached(&entity_desc, &query)
1436 .map_err(SqlExecutorError::Compile)?;
1437 let mut rows = self
1438 .transport
1439 .fetch_all_compact_sql(&compiled_readback)
1440 .await
1441 .map_err(SqlExecutorError::Transport)?;
1442 if rows.len() != 1 {
1443 return Err(SqlExecutorError::PersistedRecord(format!(
1444 "persisted {entity_name} record could not be read back"
1445 )));
1446 }
1447 rows.pop().map(|row| EntitySnapshot::from(row.into_map()))
1448 } else {
1449 None
1450 }
1451 } else {
1452 None
1453 };
1454
1455 let metadata = ExecutionMetadata {
1456 backend: "sql".to_string(),
1457 operation,
1458 started_at: start,
1459 ended_at: end,
1460 affected_rows: Some(affected_rows),
1461 result_count: None,
1462 trace_chain: request.trace_chain().to_vec(),
1463 comment: request.comment().map(|s| s.to_owned()),
1464 backend_request_id: None,
1465 parameterized_query: Some(compiled.sql.clone()),
1466 params: compiled.params.clone(),
1467 debug_query: Some(compiled.debug_sql(self.dialect.kind())),
1468 };
1469
1470 Ok(MutationResult {
1471 affected_rows,
1472 generated_values: GeneratedValues::default(),
1473 persisted_snapshot,
1474 metadata,
1475 })
1476 }
1477 }
1478}
1479
1480impl<
1481 'a,
1482 D: SqlDialect + Send + Sync,
1483 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1484 S: teaql_data_service::SchemaProvider + Send + Sync,
1485> teaql_data_service::Transaction for SqlDataServiceTransaction<'a, D, Tx, S>
1486{
1487 type Error = SqlExecutorError<<Tx as SqlTransport>::Error>;
1488
1489 fn commit(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
1490 async move {
1491 self.transport
1492 .commit_sql()
1493 .await
1494 .map_err(SqlExecutorError::Transport)
1495 }
1496 }
1497
1498 fn rollback(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
1499 async move {
1500 self.transport
1501 .rollback_sql()
1502 .await
1503 .map_err(SqlExecutorError::Transport)
1504 }
1505 }
1506}
1507
1508impl<
1509 D: SqlDialect + Send + Sync,
1510 T: SqlTransactionTransport + Send + Sync,
1511 S: teaql_data_service::SchemaProvider + Send + Sync,
1512> teaql_data_service::TransactionExecutor for SqlDataServiceExecutor<D, T, S>
1513{
1514 type Tx<'a>
1515 = SqlDataServiceTransaction<'a, D, T::Tx<'a>, S>
1516 where
1517 Self: 'a;
1518
1519 fn begin(&self) -> impl std::future::Future<Output = Result<Self::Tx<'_>, Self::Error>> + Send {
1520 async move {
1521 let tx = self
1522 .transport
1523 .begin_sql()
1524 .await
1525 .map_err(SqlExecutorError::Transport)?;
1526 Ok(SqlDataServiceTransaction {
1527 dialect: &self.dialect,
1528 transport: tx,
1529 schema_provider: &self.schema_provider,
1530 descriptor_cache: self.descriptor_cache.clone(),
1531 select_plan_cache: self.select_plan_cache.clone(),
1532 })
1533 }
1534 }
1535}
1536
1537impl<
1538 D: SqlDialect + Send + Sync,
1539 T: StreamingSqlTransport + Send + Sync,
1540 S: teaql_data_service::SchemaProvider + Send + Sync,
1541> teaql_data_service::StreamQueryExecutor for SqlDataServiceExecutor<D, T, S>
1542{
1543 fn query_stream(
1544 &self,
1545 request: teaql_data_service::QueryRequest,
1546 chunk_size: usize,
1547 ) -> teaql_data_service::QueryStream<'_, Self::Error> {
1548 use futures_util::StreamExt;
1549 let entity = match self.entity_descriptor(&request.query.entity) {
1550 Some(entity) => entity,
1551 None => {
1552 return Box::pin(futures_util::stream::once(async {
1553 Err(SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1554 request.query.entity,
1555 )))
1556 }));
1557 }
1558 };
1559 match self.compile_select_cached(&entity, &request.query) {
1560 Ok(compiled) => Box::pin(
1561 self.transport
1562 .stream_sql(compiled, chunk_size)
1563 .map(|r| r.map_err(SqlExecutorError::Transport)),
1564 ),
1565 Err(error) => Box::pin(futures_util::stream::once(async {
1566 Err(SqlExecutorError::Compile(error))
1567 })),
1568 }
1569 }
1570}