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