sql-query-connector: take trace ids by reference

query-engine/connectors/sql-query-connector/src/cursor_condition.rs

@@ -118,7 +118,7 @@ struct CursorOrderForeignKey {
 ///     OR `ModelA`.`modelB_id` IS NULL -- >>> Additional check for the nullable foreign key
 ///   )
 /// ```
-pub fn build(
+pub(crate) fn build(
     query_arguments: &QueryArguments,
     model: &ModelRef,
     order_by_defs: &[OrderByDefinition],

query-engine/connectors/sql-query-connector/src/database/connection.rs

@@ -95,7 +95,7 @@ where
                 filter,
                 &selected_fields.into(),
                 aggr_selections,
-                trace_id,
+                trace_id.as_deref(),
             )
             .await
         })
@@ -118,7 +118,7 @@ where
                 &selected_fields.into(),
                 aggr_selections,
                 SqlInfo::from(&self.connection_info),
-                trace_id,
+                trace_id.as_deref(),
             )
             .await
         })
@@ -132,7 +132,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<Vec<(SelectionResult, SelectionResult)>> {
         catch(self.connection_info.clone(), async move {
-            read::get_related_m2m_record_ids(&self.inner, from_field, from_record_ids, trace_id).await
+            read::get_related_m2m_record_ids(&self.inner, from_field, from_record_ids, trace_id.as_deref()).await
         })
         .await
     }
@@ -154,7 +154,7 @@ where
                 selections,
                 group_by,
                 having,
-                trace_id,
+                trace_id.as_deref(),
             )
             .await
         })
@@ -174,7 +174,14 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<SelectionResult> {
         catch(self.connection_info.clone(), async move {
-            write::create_record(&self.inner, &self.connection_info.sql_family(), model, args, trace_id).await
+            write::create_record(
+                &self.inner,
+                &self.connection_info.sql_family(),
+                model,
+                args,
+                trace_id.as_deref(),
+            )
+            .await
         })
         .await
     }
@@ -193,7 +200,7 @@ where
                 model,
                 args,
                 skip_duplicates,
-                trace_id,
+                trace_id.as_deref(),
             )
             .await
         })
@@ -208,7 +215,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<usize> {
         catch(self.connection_info.clone(), async move {
-            write::update_records(&self.inner, model, record_filter, args, trace_id).await
+            write::update_records(&self.inner, model, record_filter, args, trace_id.as_deref()).await
         })
         .await
     }
@@ -221,7 +228,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<Option<SelectionResult>> {
         catch(self.connection_info.clone(), async move {
-            let mut res = write::update_record(&self.inner, model, record_filter, args, trace_id).await?;
+            let mut res = write::update_record(&self.inner, model, record_filter, args, trace_id.as_deref()).await?;
             Ok(res.pop())
         })
         .await
@@ -234,7 +241,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<usize> {
         catch(self.connection_info.clone(), async move {
-            write::delete_records(&self.inner, model, record_filter, trace_id).await
+            write::delete_records(&self.inner, model, record_filter, trace_id.as_deref()).await
         })
         .await
     }
@@ -245,7 +252,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<SingleRecord> {
         catch(self.connection_info.clone(), async move {
-            native_upsert(&self.inner, upsert, trace_id).await
+            native_upsert(&self.inner, upsert, trace_id.as_deref()).await
         })
         .await
     }
@@ -270,7 +277,7 @@ where
         trace_id: Option<String>,
     ) -> connector::Result<()> {
         catch(self.connection_info.clone(), async move {
-            write::m2m_disconnect(&self.inner, field, parent_id, child_ids, trace_id).await
+            write::m2m_disconnect(&self.inner, field, parent_id, child_ids, trace_id.as_deref()).await
         })
         .await
     }

query-engine/connectors/sql-query-connector/src/database/operations/read.rs

@@ -11,21 +11,15 @@ use futures::stream::{FuturesUnordered, StreamExt};
 use prisma_models::*;
 use quaint::ast::*;
 
-pub async fn get_single_record(
+pub(crate) async fn get_single_record(
     conn: &dyn QueryExt,
     model: &ModelRef,
     filter: &Filter,
     selected_fields: &ModelProjection,
     aggr_selections: &[RelAggregationSelection],
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<Option<SingleRecord>> {
-    let query = read::get_records(
-        model,
-        selected_fields.as_columns(),
-        aggr_selections,
-        filter,
-        trace_id.clone(),
-    );
+    let query = read::get_records(model, selected_fields.as_columns(), aggr_selections, filter, trace_id);
 
     let mut field_names: Vec<_> = selected_fields.db_names().collect();
     let mut aggr_field_names: Vec<_> = aggr_selections.iter().map(|aggr_sel| aggr_sel.db_alias()).collect();
@@ -61,7 +55,7 @@ pub async fn get_many_records(
     selected_fields: &ModelProjection,
     aggr_selections: &[RelAggregationSelection],
     sql_info: SqlInfo,
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<ManyRecords> {
     let reversed = query_arguments.needs_reversed_order();
 
@@ -112,15 +106,9 @@ pub async fn get_many_records(
             let mut futures = FuturesUnordered::new();
 
             for args in batches.into_iter() {
-                let query = read::get_records(
-                    model,
-                    selected_fields.as_columns(),
-                    aggr_selections,
-                    args,
-                    trace_id.clone(),
-                );
-
-                futures.push(conn.filter(query.into(), meta.as_slice(), trace_id.clone()));
+                let query = read::get_records(model, selected_fields.as_columns(), aggr_selections, args, trace_id);
+
+                futures.push(conn.filter(query.into(), meta.as_slice(), trace_id));
             }
 
             while let Some(result) = futures.next().await {
@@ -139,7 +127,7 @@ pub async fn get_many_records(
                 selected_fields.as_columns(),
                 aggr_selections,
                 query_arguments,
-                trace_id.clone(),
+                trace_id,
             );
 
             for item in conn.filter(query.into(), meta.as_slice(), trace_id).await?.into_iter() {
@@ -159,7 +147,7 @@ pub async fn get_related_m2m_record_ids(
     conn: &dyn QueryExt,
     from_field: &RelationFieldRef,
     from_record_ids: &[SelectionResult],
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<Vec<(SelectionResult, SelectionResult)>> {
     let mut idents = vec![];
     idents.extend(ModelProjection::from(from_field.model().primary_identifier()).type_identifiers_with_arities());
@@ -231,7 +219,7 @@ pub async fn aggregate(
     selections: Vec<AggregationSelection>,
     group_by: Vec<ScalarFieldRef>,
     having: Option<Filter>,
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<Vec<AggregationRow>> {
     if !group_by.is_empty() {
         group_by_aggregate(conn, model, query_arguments, selections, group_by, having, trace_id).await
@@ -247,9 +235,9 @@ async fn plain_aggregate(
     model: &ModelRef,
     query_arguments: QueryArguments,
     selections: Vec<AggregationSelection>,
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<Vec<AggregationResult>> {
-    let query = read::aggregate(model, &selections, query_arguments, trace_id.clone());
+    let query = read::aggregate(model, &selections, query_arguments, trace_id);
 
     let idents: Vec<_> = selections
         .iter()
@@ -274,9 +262,9 @@ async fn group_by_aggregate(
     selections: Vec<AggregationSelection>,
     group_by: Vec<ScalarFieldRef>,
     having: Option<Filter>,
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<Vec<AggregationRow>> {
-    let query = read::group_by_aggregate(model, query_arguments, &selections, group_by, having, trace_id.clone());
+    let query = read::group_by_aggregate(model, query_arguments, &selections, group_by, having, trace_id);
 
     let idents: Vec<_> = selections
         .iter()

query-engine/connectors/sql-query-connector/src/database/operations/upsert.rs

@@ -1,20 +1,19 @@
-use connector_interface::NativeUpsert;
-use prisma_models::{ModelProjection, Record, SingleRecord};
-use quaint::prelude::{OnConflict, Query};
-
 use crate::{
     column_metadata,
     filter_conversion::AliasedCondition,
     model_extensions::AsColumns,
-    query_builder::{build_update_and_set_query, create_record},
+    query_builder::write::{build_update_and_set_query, create_record},
     query_ext::QueryExt,
     row::ToSqlRow,
 };
+use connector_interface::NativeUpsert;
+use prisma_models::{ModelProjection, Record, SingleRecord};
+use quaint::prelude::{OnConflict, Query};
 
-pub async fn native_upsert(
+pub(crate) async fn native_upsert(
     conn: &dyn QueryExt,
     upsert: NativeUpsert,
-    trace_id: Option<String>,
+    trace_id: Option<&str>,
 ) -> crate::Result<SingleRecord> {
     let selected_fields: ModelProjection = upsert.selected_fields().into();
     let field_names: Vec<_> = selected_fields.db_names().collect();