sql-query-connector: take trace ids by reference (#3571)

This commit is limited to sql-query-connector, it should not interfere
with #3505.

In general, we do not need ownership of the trace id in
sql-query-connector, since we only re-render it in comments. Working
with a reference is easier (it ii `Copy`, etc.), and it already saves us
string copies and allocations with this commit.

The other purpose of this PR is that we discussed yesterday about
introducing a `QueryContext<'_>` type struct to hold things like the
trace id and connection info. Experience from designing similar context
structs in the schema team showed it is much easier to do if everything
in the context can be a reference.

On the side, I could not resist making a few public items crate-public,
to make the public surface of the crate smaller and clearer.

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();