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 topn_004_011_sqlite_reuses_one_repeated_probe_boundary() {
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 topn_001_explicit_zero_threshold_forces_window_for_probe_provider() {
826 let calls = Arc::new(AtomicUsize::new(0));
827 let single_calls = Arc::new(AtomicUsize::new(0));
828 let executor = SqlDataServiceExecutor::new(
829 ProbeDialect,
830 RepeatedProbeTransport {
831 calls: calls.clone(),
832 single_calls: single_calls.clone(),
833 },
834 CountingSchemaProvider {
835 lookups: Arc::new(AtomicUsize::new(0)),
836 },
837 );
838 let mut request = query_request(false);
839 request.capture_execution_metadata = false;
840 request.query = request
841 .query
842 .filter(Expr::in_list("id", [Value::U64(7), Value::U64(9)]))
843 .order_desc("id")
844 .limit(1)
845 .partition_by("id")
846 .top_n_probe_parent_threshold(0);
847
848 executor.query(request).await.unwrap();
849
850 assert_eq!(calls.load(Ordering::Relaxed), 0);
851 assert_eq!(single_calls.load(Ordering::Relaxed), 1);
852 }
853
854 #[test]
855 fn topn_007_probe_rewrites_only_partition_membership_filter() {
856 let policy = Expr::eq("tenant_id", 7_u64);
857 let visibility = Expr::gt("version", 0_i64);
858 let business = Expr::eq("status", "ACTIVE");
859 let query = SelectQuery::new("Order")
860 .filter(Expr::and([
861 Expr::in_list("owner_id", [Value::U64(11), Value::U64(12)]),
862 policy.clone(),
863 visibility.clone(),
864 business.clone(),
865 ]))
866 .order_desc("id")
867 .limit(3)
868 .partition_by("owner_id");
869
870 let probe = scalar_partition_probe_query(&query, Value::U64(11)).unwrap();
871 let expected = Expr::and([Expr::eq("owner_id", 11_u64), policy, visibility, business]);
872
873 assert_eq!(probe.filter, Some(expected));
874 assert!(probe.partition_by.is_none());
875 assert_eq!(probe.order_by, query.order_by);
876 assert_eq!(probe.slice, query.slice);
877 }
878
879 #[tokio::test]
880 async fn cached_select_plan_rebinds_values_and_separates_in_list_lengths() {
881 let lookups = Arc::new(AtomicUsize::new(0));
882 let executor = SqlDataServiceExecutor::new(
883 TestDialect,
884 EmptyTransport,
885 CountingSchemaProvider { lookups },
886 );
887 let request = |filter| QueryRequest {
888 query: SelectQuery::new("Order").filter(filter),
889 trace_chain: Vec::new(),
890 comment: None,
891 capture_debug_query: false,
892 capture_execution_metadata: true,
893 };
894
895 let first = executor
896 .query(request(Expr::eq("id", 7_u64)))
897 .await
898 .unwrap();
899 let second = executor
900 .query(request(Expr::eq("id", 9_u64)))
901 .await
902 .unwrap();
903 assert_eq!(
904 first.metadata.parameterized_query,
905 second.metadata.parameterized_query
906 );
907 assert_eq!(first.metadata.params, vec![Value::U64(7)]);
908 assert_eq!(second.metadata.params, vec![Value::U64(9)]);
909
910 let short = executor
911 .query(request(Expr::in_list("id", [Value::U64(1), Value::U64(2)])))
912 .await
913 .unwrap();
914 let long = executor
915 .query(request(Expr::in_list(
916 "id",
917 [Value::U64(1), Value::U64(2), Value::U64(3)],
918 )))
919 .await
920 .unwrap();
921 assert_ne!(
922 short.metadata.parameterized_query,
923 long.metadata.parameterized_query
924 );
925 assert_eq!(short.metadata.params.len(), 2);
926 assert_eq!(long.metadata.params.len(), 3);
927 }
928
929 #[tokio::test]
930 async fn cached_select_plan_rebinds_large_in_as_one_array_parameter() {
931 let executor = SqlDataServiceExecutor::new(
932 ArrayTestDialect,
933 EmptyTransport,
934 CountingSchemaProvider {
935 lookups: Arc::new(AtomicUsize::new(0)),
936 },
937 );
938 let request = |values: Vec<Value>| QueryRequest {
939 query: SelectQuery::new("Order").filter(Expr::in_list("id", values)),
940 trace_chain: Vec::new(),
941 comment: None,
942 capture_debug_query: false,
943 capture_execution_metadata: true,
944 };
945 let first_values = (1_u64..=21).map(Value::from).collect::<Vec<_>>();
946 let second_values = (101_u64..=121).map(Value::from).collect::<Vec<_>>();
947
948 let first = executor.query(request(first_values.clone())).await.unwrap();
949 let second = executor
950 .query(request(second_values.clone()))
951 .await
952 .unwrap();
953
954 assert_eq!(
955 first.metadata.parameterized_query,
956 second.metadata.parameterized_query
957 );
958 assert_eq!(first.metadata.params, vec![Value::List(first_values)]);
959 assert_eq!(second.metadata.params, vec![Value::List(second_values)]);
960 }
961
962 #[tokio::test]
963 async fn cached_select_plan_preserves_parameter_order_for_supported_query_shapes() {
964 let executor = SqlDataServiceExecutor::new(
965 TestDialect,
966 EmptyTransport,
967 CountingSchemaProvider {
968 lookups: Arc::new(AtomicUsize::new(0)),
969 },
970 );
971
972 async fn assert_rebound(
973 executor: &SqlDataServiceExecutor<TestDialect, EmptyTransport, CountingSchemaProvider>,
974 warm: SelectQuery,
975 current: SelectQuery,
976 ) {
977 let request = |query| QueryRequest {
978 query,
979 trace_chain: Vec::new(),
980 comment: None,
981 capture_debug_query: false,
982 capture_execution_metadata: true,
983 };
984 executor.query(request(warm)).await.unwrap();
985 let actual = executor.query(request(current.clone())).await.unwrap();
986 let expected = TestDialect
987 .compile_select(&test_entity(), ¤t)
988 .unwrap();
989 assert_eq!(actual.metadata.parameterized_query, Some(expected.sql));
990 assert_eq!(actual.metadata.params, expected.params);
991 }
992
993 assert_rebound(
994 &executor,
995 SelectQuery::new("Order").search_with_text("first"),
996 SelectQuery::new("Order").search_with_text("second"),
997 )
998 .await;
999 assert_rebound(
1000 &executor,
1001 SelectQuery::new("Order")
1002 .project_expr("marker", Expr::value(1_i64))
1003 .filter(Expr::eq("id", 2_u64))
1004 .having(Expr::gt("id", 3_u64))
1005 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(4_i64))),
1006 SelectQuery::new("Order")
1007 .project_expr("marker", Expr::value(11_i64))
1008 .filter(Expr::eq("id", 12_u64))
1009 .having(Expr::gt("id", 13_u64))
1010 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(14_i64))),
1011 )
1012 .await;
1013 assert_rebound(
1014 &executor,
1015 SelectQuery::new("Order")
1016 .filter(Expr::eq("id", 1_u64))
1017 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(2_i64)))
1018 .page(0, 10)
1019 .partition_by("name"),
1020 SelectQuery::new("Order")
1021 .filter(Expr::eq("id", 3_u64))
1022 .order_by(teaql_core::OrderBy::asc_expr(Expr::value(4_i64)))
1023 .page(0, 10)
1024 .partition_by("name"),
1025 )
1026 .await;
1027 assert_rebound(
1028 &executor,
1029 SelectQuery::new("Order").filter(Expr::in_subquery(
1030 "id",
1031 test_entity(),
1032 SelectQuery::new("Order").filter(Expr::gt("id", 20_u64)),
1033 "id",
1034 )),
1035 SelectQuery::new("Order").filter(Expr::in_subquery(
1036 "id",
1037 test_entity(),
1038 SelectQuery::new("Order").filter(Expr::gt("id", 30_u64)),
1039 "id",
1040 )),
1041 )
1042 .await;
1043 }
1044}
1045
1046impl<
1047 D: SqlDialect + Send + Sync,
1048 T: SqlTransport + Send + Sync,
1049 S: teaql_data_service::SchemaProvider + Send + Sync,
1050> QueryExecutor for SqlDataServiceExecutor<D, T, S>
1051{
1052 fn query(
1053 &self,
1054 request: QueryRequest,
1055 ) -> impl std::future::Future<Output = Result<QueryResult, Self::Error>> + Send {
1056 async move {
1057 let entity_desc = self
1058 .entity_descriptor(&request.query.entity)
1059 .ok_or_else(|| {
1060 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1061 request.query.entity.clone(),
1062 ))
1063 })?;
1064
1065 if !request.capture_execution_metadata
1066 && self.dialect.prefers_small_parent_relation_probes()
1067 && request.query.top_n_probe_parent_threshold.is_none()
1068 && let Some(values) = partition_probe_values(&request.query)
1069 && values.len() >= 2
1070 && let (Some(first_query), Some(second_query)) = (
1071 scalar_partition_probe_query(&request.query, values[0].clone()),
1072 scalar_partition_probe_query(&request.query, values[1].clone()),
1073 )
1074 {
1075 let first = self
1076 .compile_select_cached(&entity_desc, &first_query)
1077 .map_err(SqlExecutorError::Compile)?;
1078 let second_params = collect_select_params(
1079 &entity_desc,
1080 &second_query,
1081 self.dialect.large_in_uses_array_param(),
1082 );
1083 if let Some(param_index) = first
1084 .params
1085 .iter()
1086 .zip(&second_params)
1087 .position(|(left, right)| left != right)
1088 {
1089 let rows = self
1090 .transport
1091 .fetch_repeated_compact_sql(&first, param_index, &values)
1092 .await
1093 .map_err(SqlExecutorError::Transport)?;
1094 return Ok(QueryResult {
1095 metadata: ExecutionMetadata::unrecorded_query(rows.len()),
1096 rows,
1097 });
1098 }
1099 }
1100
1101 let compiled = self
1102 .compile_select_cached(&entity_desc, &request.query)
1103 .map_err(SqlExecutorError::Compile)?;
1104 let start = request.capture_execution_metadata.then(SystemTime::now);
1105 let rows = self
1106 .transport
1107 .fetch_all_compact_sql(&compiled)
1108 .await
1109 .map_err(SqlExecutorError::Transport)?;
1110 let end = request.capture_execution_metadata.then(SystemTime::now);
1111 let debug_query = request
1112 .capture_debug_query
1113 .then(|| compiled.debug_sql(self.dialect.kind()));
1114 let metadata = if request.capture_execution_metadata {
1115 let CompiledQuery { sql, params, .. } = compiled;
1116 ExecutionMetadata {
1117 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1118 operation: DataServiceOperation::Query,
1119 started_at: start.expect("captured query start"),
1120 ended_at: end.expect("captured query end"),
1121 affected_rows: None,
1122 result_count: Some(rows.len()),
1123 trace_chain: request.trace_chain,
1124 comment: request.comment,
1125 backend_request_id: None,
1126 parameterized_query: Some(sql),
1127 params,
1128 debug_query,
1129 }
1130 } else {
1131 ExecutionMetadata::unrecorded_query(rows.len())
1132 };
1133
1134 Ok(QueryResult { rows, metadata })
1135 }
1136 }
1137}
1138
1139impl<
1140 D: SqlDialect + Send + Sync,
1141 T: SqlTransport + Send + Sync,
1142 S: teaql_data_service::SchemaProvider + Send + Sync,
1143> MutationExecutor for SqlDataServiceExecutor<D, T, S>
1144{
1145 fn mutate(
1146 &self,
1147 request: MutationRequest,
1148 ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send {
1149 async move {
1150 let entity_name = match &request {
1151 MutationRequest::Insert(cmd) => &cmd.entity,
1152 MutationRequest::Update(cmd) => &cmd.entity,
1153 MutationRequest::Delete(cmd) => &cmd.entity,
1154 MutationRequest::Recover(cmd) => &cmd.entity,
1155 MutationRequest::Batch(mutations) => {
1156 let mut total_affected = 0;
1157 let mut parameterized_queries = Vec::new();
1158 let mut params = Vec::new();
1159 let mut debug_queries = Vec::new();
1160 let start = SystemTime::now();
1161 for req in mutations {
1162 let res = Box::pin(self.mutate(req.clone())).await?;
1163 total_affected += res.affected_rows;
1164 if let Some(query) = res.metadata.parameterized_query {
1165 parameterized_queries.push(query);
1166 }
1167 params.extend(res.metadata.params);
1168 if let Some(query) = res.metadata.debug_query {
1169 debug_queries.push(query);
1170 }
1171 }
1172 let end = SystemTime::now();
1173 return Ok(MutationResult {
1174 affected_rows: total_affected,
1175 generated_values: GeneratedValues::default(),
1176 persisted_snapshot: None,
1177 metadata: ExecutionMetadata {
1178 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1179 operation: DataServiceOperation::Batch,
1180 started_at: start,
1181 ended_at: end,
1182 affected_rows: Some(total_affected),
1183 result_count: None,
1184 trace_chain: Vec::new(),
1185 comment: None,
1186 backend_request_id: None,
1187 parameterized_query: (!parameterized_queries.is_empty())
1188 .then(|| parameterized_queries.join("; ")),
1189 params,
1190 debug_query: (!debug_queries.is_empty())
1191 .then(|| debug_queries.join("; ")),
1192 },
1193 });
1194 }
1195 };
1196
1197 let entity_desc = self.entity_descriptor(entity_name).ok_or_else(|| {
1198 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(entity_name.clone()))
1199 })?;
1200
1201 let compiled = match &request {
1202 MutationRequest::Insert(cmd) => self
1203 .dialect
1204 .compile_insert(&entity_desc, cmd)
1205 .map_err(SqlExecutorError::Compile)?,
1206 MutationRequest::Update(cmd) => self
1207 .dialect
1208 .compile_update(&entity_desc, cmd)
1209 .map_err(SqlExecutorError::Compile)?,
1210 MutationRequest::Delete(cmd) => self
1211 .dialect
1212 .compile_delete(&entity_desc, cmd)
1213 .map_err(SqlExecutorError::Compile)?,
1214 MutationRequest::Recover(cmd) => self
1215 .dialect
1216 .compile_recover(&entity_desc, cmd)
1217 .map_err(SqlExecutorError::Compile)?,
1218 MutationRequest::Batch(_) => unreachable!(),
1219 };
1220
1221 let start = SystemTime::now();
1222 let affected_rows = self
1223 .transport
1224 .execute_sql(&compiled)
1225 .await
1226 .map_err(SqlExecutorError::Transport)?;
1227 let end = SystemTime::now();
1228
1229 let operation = match &request {
1230 MutationRequest::Insert(_) => DataServiceOperation::Insert,
1231 MutationRequest::Update(_) => DataServiceOperation::Update,
1232 MutationRequest::Delete(_) => DataServiceOperation::Delete,
1233 MutationRequest::Recover(_) => DataServiceOperation::Recover,
1234 MutationRequest::Batch(_) => DataServiceOperation::Batch,
1235 };
1236
1237 let metadata = ExecutionMetadata {
1238 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1239 operation,
1240 started_at: start,
1241 ended_at: end,
1242 affected_rows: Some(affected_rows),
1243 result_count: None,
1244 trace_chain: request.trace_chain().to_vec(),
1245 comment: request.comment().map(|s| s.to_owned()),
1246 backend_request_id: None,
1247 parameterized_query: Some(compiled.sql.clone()),
1248 params: compiled.params.clone(),
1249 debug_query: Some(compiled.debug_sql(self.dialect.kind())),
1250 };
1251
1252 Ok(MutationResult {
1253 affected_rows,
1254 generated_values: GeneratedValues::default(),
1255 persisted_snapshot: None,
1256 metadata,
1257 })
1258 }
1259 }
1260}
1261
1262#[derive(Clone)]
1263pub struct SqlDataServiceTransaction<'a, D, Tx: SqlTransport + SqlTransaction, S> {
1264 pub dialect: &'a D,
1265 pub transport: Tx,
1266 pub schema_provider: &'a S,
1267 descriptor_cache: Arc<RwLock<HashMap<String, Arc<teaql_core::EntityDescriptor>>>>,
1268 select_plan_cache: Arc<RwLock<Vec<(SelectQuery, String)>>>,
1269}
1270
1271impl<'a, D, Tx: SqlTransport + SqlTransaction, S> SqlDataServiceTransaction<'a, D, Tx, S>
1272where
1273 S: teaql_data_service::SchemaProvider,
1274{
1275 fn entity_descriptor(&self, name: &str) -> Option<Arc<teaql_core::EntityDescriptor>> {
1276 if let Ok(cache) = self.descriptor_cache.read()
1277 && let Some(descriptor) = cache.get(name)
1278 {
1279 return Some(descriptor.clone());
1280 }
1281 let descriptor = self.schema_provider.get_entity(name)?;
1282 if let Ok(mut cache) = self.descriptor_cache.write() {
1283 return Some(
1284 cache
1285 .entry(name.to_owned())
1286 .or_insert_with(|| descriptor.clone())
1287 .clone(),
1288 );
1289 }
1290 Some(descriptor)
1291 }
1292
1293 fn compile_select_cached(
1294 &self,
1295 entity: &EntityDescriptor,
1296 query: &SelectQuery,
1297 ) -> Result<CompiledQuery, SqlCompileError>
1298 where
1299 D: SqlDialect,
1300 {
1301 compile_select_with_cache(self.dialect, &self.select_plan_cache, entity, query)
1302 }
1303}
1304
1305impl<
1306 'a,
1307 D: SqlDialect + Send + Sync,
1308 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1309 S: teaql_data_service::SchemaProvider + Send + Sync,
1310> DataServiceExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1311{
1312 type Error = SqlExecutorError<<Tx as SqlTransport>::Error>;
1313
1314 fn capabilities(&self) -> DataServiceCapabilities {
1315 DataServiceCapabilities {
1316 query: true,
1317 mutation: true,
1318 transaction: false,
1319 schema: false,
1320 id_generation: false,
1321 batch_mutation: true,
1322 returning: false,
1323 small_parent_relation_probes: self.dialect.prefers_small_parent_relation_probes(),
1324 }
1325 }
1326}
1327
1328impl<
1329 'a,
1330 D: SqlDialect + Send + Sync,
1331 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1332 S: teaql_data_service::SchemaProvider + Send + Sync,
1333> QueryExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1334{
1335 fn query(
1336 &self,
1337 request: QueryRequest,
1338 ) -> impl std::future::Future<Output = Result<QueryResult, Self::Error>> + Send {
1339 async move {
1340 let entity_desc = self
1341 .entity_descriptor(&request.query.entity)
1342 .ok_or_else(|| {
1343 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1344 request.query.entity.clone(),
1345 ))
1346 })?;
1347
1348 let compiled = self
1349 .compile_select_cached(&entity_desc, &request.query)
1350 .map_err(SqlExecutorError::Compile)?;
1351 let start = SystemTime::now();
1352 let rows = self
1353 .transport
1354 .fetch_all_compact_sql(&compiled)
1355 .await
1356 .map_err(SqlExecutorError::Transport)?;
1357 let end = SystemTime::now();
1358
1359 let metadata = ExecutionMetadata {
1360 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1361 operation: DataServiceOperation::Query,
1362 started_at: start,
1363 ended_at: end,
1364 affected_rows: None,
1365 result_count: Some(rows.len()),
1366 trace_chain: request.trace_chain,
1367 comment: request.comment,
1368 backend_request_id: None,
1369 parameterized_query: Some(compiled.sql.clone()),
1370 params: compiled.params.clone(),
1371 debug_query: request
1372 .capture_debug_query
1373 .then(|| compiled.debug_sql(self.dialect.kind())),
1374 };
1375
1376 Ok(QueryResult { rows, metadata })
1377 }
1378 }
1379}
1380
1381impl<
1382 'a,
1383 D: SqlDialect + Send + Sync,
1384 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1385 S: teaql_data_service::SchemaProvider + Send + Sync,
1386> teaql_data_service::StreamQueryExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1387{
1388 fn query_stream(
1389 &self,
1390 request: QueryRequest,
1391 chunk_size: usize,
1392 ) -> teaql_data_service::QueryStream<'_, Self::Error> {
1393 use std::collections::VecDeque;
1394
1395 let chunk_size = chunk_size.max(1);
1396 Box::pin(futures_util::stream::try_unfold(
1397 (Some(request), VecDeque::new(), 0_usize),
1398 move |(request, mut rows, chunk_index)| async move {
1399 if let Some(request) = request {
1400 rows = QueryExecutor::query(self, request).await?.rows.into();
1401 }
1402 if rows.is_empty() {
1403 return Ok(None);
1404 }
1405 let take = rows.len().min(chunk_size);
1406 let chunk_rows = rows.drain(..take).collect();
1407 let is_last = rows.is_empty();
1408 Ok(Some((
1409 teaql_data_service::StreamChunk {
1410 rows: chunk_rows,
1411 chunk_index,
1412 is_last,
1413 },
1414 (None, rows, chunk_index + 1),
1415 )))
1416 },
1417 ))
1418 }
1419}
1420
1421impl<
1422 'a,
1423 D: SqlDialect + Send + Sync,
1424 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1425 S: teaql_data_service::SchemaProvider + Send + Sync,
1426> MutationExecutor for SqlDataServiceTransaction<'a, D, Tx, S>
1427{
1428 fn mutate(
1429 &self,
1430 request: MutationRequest,
1431 ) -> impl std::future::Future<Output = Result<MutationResult, Self::Error>> + Send {
1432 async move {
1433 let entity_name = match &request {
1434 MutationRequest::Insert(cmd) => &cmd.entity,
1435 MutationRequest::Update(cmd) => &cmd.entity,
1436 MutationRequest::Delete(cmd) => &cmd.entity,
1437 MutationRequest::Recover(cmd) => &cmd.entity,
1438 MutationRequest::Batch(mutations) => {
1439 let mut total_affected = 0;
1440 let mut parameterized_queries = Vec::new();
1441 let mut params = Vec::new();
1442 let mut debug_queries = Vec::new();
1443 let start = SystemTime::now();
1444 for req in mutations {
1445 let res = Box::pin(self.mutate(req.clone())).await?;
1446 total_affected += res.affected_rows;
1447 if let Some(query) = res.metadata.parameterized_query {
1448 parameterized_queries.push(query);
1449 }
1450 params.extend(res.metadata.params);
1451 if let Some(query) = res.metadata.debug_query {
1452 debug_queries.push(query);
1453 }
1454 }
1455 let end = SystemTime::now();
1456 return Ok(MutationResult {
1457 affected_rows: total_affected,
1458 generated_values: GeneratedValues::default(),
1459 persisted_snapshot: None,
1460 metadata: ExecutionMetadata {
1461 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1462 operation: DataServiceOperation::Batch,
1463 started_at: start,
1464 ended_at: end,
1465 affected_rows: Some(total_affected),
1466 result_count: None,
1467 trace_chain: Vec::new(),
1468 comment: None,
1469 backend_request_id: None,
1470 parameterized_query: (!parameterized_queries.is_empty())
1471 .then(|| parameterized_queries.join("; ")),
1472 params,
1473 debug_query: (!debug_queries.is_empty())
1474 .then(|| debug_queries.join("; ")),
1475 },
1476 });
1477 }
1478 };
1479
1480 let entity_desc = self.entity_descriptor(entity_name).ok_or_else(|| {
1481 SqlExecutorError::Compile(SqlCompileError::UnknownEntity(entity_name.clone()))
1482 })?;
1483
1484 let compiled = match &request {
1485 MutationRequest::Insert(cmd) => self
1486 .dialect
1487 .compile_insert(&entity_desc, cmd)
1488 .map_err(SqlExecutorError::Compile)?,
1489 MutationRequest::Update(cmd) => self
1490 .dialect
1491 .compile_update(&entity_desc, cmd)
1492 .map_err(SqlExecutorError::Compile)?,
1493 MutationRequest::Delete(cmd) => self
1494 .dialect
1495 .compile_delete(&entity_desc, cmd)
1496 .map_err(SqlExecutorError::Compile)?,
1497 MutationRequest::Recover(cmd) => self
1498 .dialect
1499 .compile_recover(&entity_desc, cmd)
1500 .map_err(SqlExecutorError::Compile)?,
1501 MutationRequest::Batch(_) => unreachable!("batch handled above"),
1502 };
1503
1504 let start = SystemTime::now();
1505 let affected_rows = self
1506 .transport
1507 .execute_sql(&compiled)
1508 .await
1509 .map_err(SqlExecutorError::Transport)?;
1510 let end = SystemTime::now();
1511
1512 let operation = match &request {
1513 MutationRequest::Insert(_) => DataServiceOperation::Insert,
1514 MutationRequest::Update(_) => DataServiceOperation::Update,
1515 MutationRequest::Delete(_) => DataServiceOperation::Delete,
1516 MutationRequest::Recover(_) => DataServiceOperation::Recover,
1517 MutationRequest::Batch(_) => DataServiceOperation::Batch,
1518 };
1519
1520 let persisted_id = match &request {
1521 MutationRequest::Insert(cmd) => cmd.values.get("id").cloned(),
1522 MutationRequest::Update(cmd) => Some(cmd.id.clone()),
1523 MutationRequest::Delete(cmd) if cmd.soft_delete => Some(cmd.id.clone()),
1524 MutationRequest::Recover(cmd) => Some(cmd.id.clone()),
1525 MutationRequest::Delete(_) | MutationRequest::Batch(_) => None,
1526 };
1527 let persisted_snapshot = if affected_rows == 1 {
1528 if let Some(id) = persisted_id {
1529 let query = SelectQuery::new(entity_name.clone()).filter(Expr::eq("id", id));
1530 let compiled_readback = self
1531 .compile_select_cached(&entity_desc, &query)
1532 .map_err(SqlExecutorError::Compile)?;
1533 let mut rows = self
1534 .transport
1535 .fetch_all_compact_sql(&compiled_readback)
1536 .await
1537 .map_err(SqlExecutorError::Transport)?;
1538 if rows.len() != 1 {
1539 return Err(SqlExecutorError::PersistedRecord(format!(
1540 "persisted {entity_name} record could not be read back"
1541 )));
1542 }
1543 rows.pop().map(|row| EntitySnapshot::from(row.into_map()))
1544 } else {
1545 None
1546 }
1547 } else {
1548 None
1549 };
1550
1551 let metadata = ExecutionMetadata {
1552 backend: format!("{:?}", self.dialect.kind()).to_ascii_lowercase(),
1553 operation,
1554 started_at: start,
1555 ended_at: end,
1556 affected_rows: Some(affected_rows),
1557 result_count: None,
1558 trace_chain: request.trace_chain().to_vec(),
1559 comment: request.comment().map(|s| s.to_owned()),
1560 backend_request_id: None,
1561 parameterized_query: Some(compiled.sql.clone()),
1562 params: compiled.params.clone(),
1563 debug_query: Some(compiled.debug_sql(self.dialect.kind())),
1564 };
1565
1566 Ok(MutationResult {
1567 affected_rows,
1568 generated_values: GeneratedValues::default(),
1569 persisted_snapshot,
1570 metadata,
1571 })
1572 }
1573 }
1574}
1575
1576impl<
1577 'a,
1578 D: SqlDialect + Send + Sync,
1579 Tx: SqlTransport + SqlTransaction<Error = <Tx as SqlTransport>::Error> + Send + Sync,
1580 S: teaql_data_service::SchemaProvider + Send + Sync,
1581> teaql_data_service::Transaction for SqlDataServiceTransaction<'a, D, Tx, S>
1582{
1583 type Error = SqlExecutorError<<Tx as SqlTransport>::Error>;
1584
1585 fn commit(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
1586 async move {
1587 self.transport
1588 .commit_sql()
1589 .await
1590 .map_err(SqlExecutorError::Transport)
1591 }
1592 }
1593
1594 fn rollback(self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
1595 async move {
1596 self.transport
1597 .rollback_sql()
1598 .await
1599 .map_err(SqlExecutorError::Transport)
1600 }
1601 }
1602}
1603
1604impl<
1605 D: SqlDialect + Send + Sync,
1606 T: SqlTransactionTransport + Send + Sync,
1607 S: teaql_data_service::SchemaProvider + Send + Sync,
1608> teaql_data_service::TransactionExecutor for SqlDataServiceExecutor<D, T, S>
1609{
1610 type Tx<'a>
1611 = SqlDataServiceTransaction<'a, D, T::Tx<'a>, S>
1612 where
1613 Self: 'a;
1614
1615 fn begin(&self) -> impl std::future::Future<Output = Result<Self::Tx<'_>, Self::Error>> + Send {
1616 async move {
1617 let tx = self
1618 .transport
1619 .begin_sql()
1620 .await
1621 .map_err(SqlExecutorError::Transport)?;
1622 Ok(SqlDataServiceTransaction {
1623 dialect: &self.dialect,
1624 transport: tx,
1625 schema_provider: &self.schema_provider,
1626 descriptor_cache: self.descriptor_cache.clone(),
1627 select_plan_cache: self.select_plan_cache.clone(),
1628 })
1629 }
1630 }
1631}
1632
1633impl<
1634 D: SqlDialect + Send + Sync,
1635 T: StreamingSqlTransport + Send + Sync,
1636 S: teaql_data_service::SchemaProvider + Send + Sync,
1637> teaql_data_service::StreamQueryExecutor for SqlDataServiceExecutor<D, T, S>
1638{
1639 fn query_stream(
1640 &self,
1641 request: teaql_data_service::QueryRequest,
1642 chunk_size: usize,
1643 ) -> teaql_data_service::QueryStream<'_, Self::Error> {
1644 use futures_util::StreamExt;
1645 let entity = match self.entity_descriptor(&request.query.entity) {
1646 Some(entity) => entity,
1647 None => {
1648 return Box::pin(futures_util::stream::once(async {
1649 Err(SqlExecutorError::Compile(SqlCompileError::UnknownEntity(
1650 request.query.entity,
1651 )))
1652 }));
1653 }
1654 };
1655 match self.compile_select_cached(&entity, &request.query) {
1656 Ok(compiled) => Box::pin(
1657 self.transport
1658 .stream_sql(compiled, chunk_size)
1659 .map(|r| r.map_err(SqlExecutorError::Transport)),
1660 ),
1661 Err(error) => Box::pin(futures_util::stream::once(async {
1662 Err(SqlExecutorError::Compile(error))
1663 })),
1664 }
1665 }
1666}