Stored procedures execution (Incremental PR - II) (#141)

<!-- The PR description should answer 2 (maybe 3) important questions:
-->

### What

<!-- What is this PR trying to accomplish (and why, if it's not
obvious)? -->

This PR is the second incremental PR towards adding stored procedures
support and this PR adds the execution logic of the same. The actual PR
logic is not very large but due to the tests, the lines diff is around
18k.

### How

<!-- How is it trying to accomplish it (what are the implementation
steps)? -->

Steps:

1. Identify whether the procedure name in the mutation request is a
stored procedure.
2. Fetch the metadata of the stored procedure.
3. Arguments and field validation, throw error if any required argument
is not provided in the request.
4. Create a temporary table, the schema of the temporary table is
determined by the `returns` field of the stored procedure. The temporary
table is used to write the results of the stored procedure into.
5. The SQL query is generated in the following format:

```sql
INSERT INTO #TempTable (col1, col2) EXEC <stored_procedure_name> @arg1_name = arg_value 
```

6. Make another SQL query to query from the temporary table with the
fields requested and return the response.

README.md

@@ -101,7 +101,6 @@ MY_SUBGRAPH_MY_SQL_WRITE_URL=http://local.hasura.dev:8081
 ### 5. Start the connector's docker compose
 
 Let's start our connector's docker compose file. Run the following from the connector's subdirectory inside a subgraph:
-
 ```bash title="Run the following from the connector's subdirectory inside a subgraph:"
 docker compose -f docker-compose.my_sql.yaml up
 ```

crates/ndc-sqlserver/tests/goldenfiles/mutations/stored_procedures/get_customer_details_with_invoices.json

@@ -0,0 +1,52 @@
+{
+  "operations": [
+    {
+      "type": "procedure",
+      "name": "GetCustomerDetailsWithTotalPurchases",
+      "arguments": {
+        "CustomerId": 1,
+        "Phone": "123"
+      },
+      "fields": {
+        "type": "array",
+        "fields": {
+          "type": "object",
+          "fields": {
+            "CustomerId": {
+              "type": "column",
+              "column": "CustomerId"
+            },
+            "Invoices": {
+              "type": "relationship",
+              "arguments": {},
+              "column": "CustomerId",
+              "relationship": "GetInvoice",
+              "query": {
+                "fields": {
+                  "InvoiceId": {
+                    "type": "column",
+                    "column": "InvoiceId"
+                  },
+                  "Total": {
+                    "type": "column",
+                    "column": "Total"
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  ],
+  "collection_relationships": {
+    "GetInvoice": {
+      "column_mapping": {
+        "CustomerId": "CustomerId"
+      },
+      "relationship_type": "array",
+      "target_collection": "Invoice",
+      "arguments": {}
+    }
+  }
+}

crates/ndc-sqlserver/tests/goldenfiles/mutations/stored_procedures/get_customer_details_without_arguments.json

@@ -0,0 +1,22 @@
+{
+    "operations": [
+        {
+          "type": "procedure",
+          "name": "GetCustomerDetailsWithTotalPurchases",
+          "arguments": {},
+          "fields": {
+            "type": "array",
+            "fields": {
+              "type": "object",
+              "fields": {
+                "CustomerId": {
+                  "type": "column",
+                  "column": "CustomerId"
+                }
+              }
+            }
+          }
+        }
+    ],
+  "collection_relationships": {}
+}

crates/ndc-sqlserver/tests/goldenfiles/mutations/stored_procedures/return_one.json

@@ -0,0 +1,22 @@
+{
+    "operations": [
+        {
+          "type": "procedure",
+          "name": "ReturnOne",
+          "arguments": {},
+          "fields": {
+            "type": "array",
+            "fields": {
+              "type": "object",
+              "fields": {
+                "ReturnOneResult": {
+                  "type": "column",
+                  "column": "result"
+                }
+              }
+            }
+          }
+        }
+    ],
+  "collection_relationships": {}
+}

crates/ndc-sqlserver/tests/mutation_tests.rs

@@ -9,7 +9,7 @@ mod native_mutations {
     #[tokio::test(flavor = "multi_thread")]
     async fn native_mutation_insert_artist_and_return_id() {
         let original_db_config = MSSQLDatabaseConfig::original_db_config();
-        let fresh_deployment = FreshDeployment::create(original_db_config, "static", vec![])
+        let fresh_deployment = FreshDeployment::create(original_db_config, "static/tests", vec![])
             .await
             .unwrap();
 
@@ -26,7 +26,7 @@ mod native_mutations {
     /// Native mutation that selects a relationship.
     async fn native_mutation_insert_artist_and_return_artist() {
         let original_db_config = MSSQLDatabaseConfig::original_db_config();
-        let fresh_deployment = FreshDeployment::create(original_db_config, "static", vec![])
+        let fresh_deployment = FreshDeployment::create(original_db_config, "static/tests", vec![])
             .await
             .unwrap();
 
@@ -38,40 +38,170 @@ mod native_mutations {
 
         insta::assert_json_snapshot!(result);
     }
+
+    mod negative_native_mutations_test {
+        use crate::common::{
+            database::MSSQLDatabaseConfig,
+            helpers::{run_mutation_fail, run_query_with_connection_uri},
+        };
+
+        use super::*;
+
+        use hyper::StatusCode;
+
+        #[tokio::test(flavor = "multi_thread")]
+        async fn test_atomicity_native_mutations() {
+            let original_db_config = MSSQLDatabaseConfig::original_db_config();
+            let fresh_deployment =
+                FreshDeployment::create(original_db_config, "static/tests", vec![])
+                    .await
+                    .unwrap();
+
+            // Mutation that tries to insert two records with the same primary key
+            // So, the second mutation is expected to fail. Since, we run the whole
+            // thing in a transaction, we expect the whole transaction to be rolled
+            // back.
+            let _ = run_mutation_fail(
+                "fail_insert_artist_and_return_id",
+                fresh_deployment.connection_uri.clone(),
+                StatusCode::INTERNAL_SERVER_ERROR,
+            )
+            .await;
+
+            let result = run_query_with_connection_uri(
+                "fetch_artist_count",
+                &fresh_deployment.connection_uri.clone(),
+            )
+            .await;
+
+            insta::assert_json_snapshot!(result);
+        }
+    }
 }
 
-mod negative_native_mutations_test {
+mod stored_procedures {
     use crate::common::{
+        configuration::get_path_from_project_root,
         database::MSSQLDatabaseConfig,
-        helpers::{run_mutation_fail, run_query_with_connection_uri},
+        helpers::{run_mutation, run_mutation_fail},
     };
 
     use super::common::fresh_deployments::FreshDeployment;
 
     use hyper::StatusCode;
 
     #[tokio::test(flavor = "multi_thread")]
-    async fn test_atomicity_native_mutations() {
+    async fn basic_stored_procedure_execution() {
         let original_db_config = MSSQLDatabaseConfig::original_db_config();
-        let fresh_deployment = FreshDeployment::create(original_db_config, "static", vec![])
-            .await
-            .unwrap();
 
-        // Mutation that tries to insert two records with the same primary key
-        // So, the second mutation is expected to fail. Since, we run the whole
-        // thing in a transaction, we expect the whole transaction to be rolled
-        // back.
-        let _ = run_mutation_fail(
-            "fail_insert_artist_and_return_id",
+        let stored_procs_setup_file_path =
+            get_path_from_project_root("static/tests/stored_procedures_sql/return_one.sql");
+
+        let fresh_deployment = FreshDeployment::create(
+            original_db_config,
+            "static/tests",
+            vec![stored_procs_setup_file_path],
+        )
+        .await
+        .unwrap();
+
+        let result = run_mutation(
+            "mutations/stored_procedures/return_one",
+            fresh_deployment.connection_uri.clone(),
+        )
+        .await;
+
+        insta::assert_json_snapshot!(result);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn execute_stored_procedure_without_providing_arguments() {
+        let original_db_config = MSSQLDatabaseConfig::original_db_config();
+
+        let stored_procs_setup_file_path = get_path_from_project_root(
+            "static/tests/stored_procedures_sql/get_customer_details.sql",
+        );
+
+        let fresh_deployment = FreshDeployment::create(
+            original_db_config,
+            "static/tests",
+            vec![stored_procs_setup_file_path],
+        )
+        .await
+        .unwrap();
+
+        let result = run_mutation_fail(
+            "mutations/stored_procedures/get_customer_details_without_arguments",
             fresh_deployment.connection_uri.clone(),
-            StatusCode::INTERNAL_SERVER_ERROR,
+            StatusCode::BAD_REQUEST,
         )
         .await;
 
-        let result =
-            run_query_with_connection_uri("fetch_artist_count", &fresh_deployment.connection_uri)
-                .await;
+        insta::assert_json_snapshot!(result);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn execute_stored_procedure_with_relationships() {
+        let original_db_config = MSSQLDatabaseConfig::original_db_config();
+
+        let stored_procs_setup_file_path = get_path_from_project_root(
+            "static/tests/stored_procedures_sql/get_customer_details.sql",
+        );
+
+        let fresh_deployment = FreshDeployment::create(
+            original_db_config,
+            "static/tests",
+            vec![stored_procs_setup_file_path],
+        )
+        .await
+        .unwrap();
+
+        let result = run_mutation(
+            "mutations/stored_procedures/get_customer_details_with_invoices",
+            fresh_deployment.connection_uri.clone(),
+        )
+        .await;
 
         insta::assert_json_snapshot!(result);
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn execute_stored_procedures_concurrently() {
+        let original_db_config = MSSQLDatabaseConfig::original_db_config();
+
+        let stored_procs_setup_file_path = get_path_from_project_root(
+            "static/tests/stored_procedures_sql/get_customer_details.sql",
+        );
+
+        let fresh_deployment = FreshDeployment::create(
+            original_db_config,
+            "static/tests",
+            vec![stored_procs_setup_file_path],
+        )
+        .await
+        .unwrap();
+
+        let fresh_deployment_1 = fresh_deployment.connection_uri.clone();
+
+        let result1 = tokio::spawn(async move {
+            let connection_uri = fresh_deployment_1.clone();
+
+            run_mutation(
+                "mutations/stored_procedures/get_customer_details_with_invoices",
+                connection_uri.clone(),
+            )
+            .await
+        });
+
+        let result2 = tokio::spawn(async move {
+            run_mutation(
+                "mutations/stored_procedures/get_customer_details_with_invoices",
+                fresh_deployment.connection_uri.clone(),
+            )
+            .await
+        });
+
+        // If either of the results are `Err`, then this will panic and the test will fail.
+        let (_result1, _result2) = tokio::try_join!(result1, result2).unwrap();
+    }
 }

crates/ndc-sqlserver/tests/snapshots/mutation_tests__native_mutations__negative_native_mutations_test__atomicity_native_mutations.snap

@@ -0,0 +1,11 @@
+---
+source: crates/ndc-sqlserver/tests/mutation_tests.rs
+expression: result
+---
+[
+  {
+    "aggregates": {
+      "artist_count": 275
+    }
+  }
+]

crates/ndc-sqlserver/tests/snapshots/mutation_tests__stored_procedures__basic_stored_procedure_execution.snap

@@ -0,0 +1,16 @@
+---
+source: crates/ndc-sqlserver/tests/mutation_tests.rs
+expression: result
+---
+{
+  "operation_results": [
+    {
+      "type": "procedure",
+      "result": [
+        {
+          "ReturnOneResult": 1
+        }
+      ]
+    }
+  ]
+}

crates/ndc-sqlserver/tests/snapshots/mutation_tests__stored_procedures__execute_stored_procedure_with_relationships.snap

@@ -0,0 +1,48 @@
+---
+source: crates/ndc-sqlserver/tests/mutation_tests.rs
+expression: result
+---
+{
+  "operation_results": [
+    {
+      "type": "procedure",
+      "result": [
+        {
+          "CustomerId": 1,
+          "Invoices": {
+            "rows": [
+              {
+                "InvoiceId": 98,
+                "Total": 3.98
+              },
+              {
+                "InvoiceId": 121,
+                "Total": 3.96
+              },
+              {
+                "InvoiceId": 143,
+                "Total": 5.94
+              },
+              {
+                "InvoiceId": 195,
+                "Total": 0.99
+              },
+              {
+                "InvoiceId": 316,
+                "Total": 1.98
+              },
+              {
+                "InvoiceId": 327,
+                "Total": 13.86
+              },
+              {
+                "InvoiceId": 382,
+                "Total": 8.91
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}