WIP

Author: danieljharvey

crates/configuration/src/connect.rs

@@ -18,6 +18,12 @@ pub fn get_connect_options(
         ConnectionUri(Secret::FromEnvironment { variable }) => {
             Cow::Owned(environment.read(variable)?)
         }
+        ConnectionUri(Secret::FromEnvironmentMap { variable, map_key }) => {
+            let map_str = environment.read(variable)?;
+            let parsed: serde_json::Value = serde_json::from_str(&map_str)?;
+            let obj = parsed.as_object().unwrap();
+            Cow::Owned(obj.get(map_key).unwrap().to_string())
+        }
     };
 
     let connect_options = PgConnectOptions::from_url(&uri.parse()?)?;

crates/configuration/src/values/secret.rs

@@ -9,7 +9,13 @@ use crate::environment;
 #[serde(untagged, rename_all = "camelCase")]
 pub enum Secret {
     Plain(String),
-    FromEnvironment { variable: environment::Variable },
+    FromEnvironment {
+        variable: environment::Variable,
+    },
+    FromEnvironmentMap {
+        variable: environment::Variable,
+        map_key: String,
+    },
 }
 
 // This conversion is useful for testing.

crates/connectors/ndc-postgres/src/query.rs

@@ -6,6 +6,8 @@
 mod explain;
 pub use explain::explain;
 
+use ndc_postgres_configuration::version5::DynamicConnectionSettings;
+use ndc_postgres_configuration::ConnectionSettings;
 use tracing::{info_span, Instrument};
 
 use ndc_sdk::connector;
@@ -38,6 +40,9 @@ pub async fn query(
             query_request = ?query_request
         );
 
+        let connection_details =
+            extract_connection_details_from_query(configuration, &query_request).await?;
+
         let plan = async {
             plan_query(configuration, state, query_request).map_err(|err| {
                 record::translation_error(&err, &state.query_metrics);
@@ -48,10 +53,12 @@ pub async fn query(
         .await?;
 
         let result = async {
-            execute_query(state, plan).await.map_err(|err| {
-                record::execution_error(&err, &state.query_metrics);
-                convert::execution_error_to_response(err)
-            })
+            execute_query(state, plan, connection_details)
+                .await
+                .map_err(|err| {
+                    record::execution_error(&err, &state.query_metrics);
+                    convert::execution_error_to_response(err)
+                })
         }
         .instrument(info_span!("Execute query"))
         .await?;
@@ -65,6 +72,44 @@ pub async fn query(
     timer.complete_with(result)
 }
 
+// if this is a dynamic connection situation, which connection string should we be using?
+pub struct RequestConnectionDetails {
+    pool: PgPool 
+}
+
+async fn extract_connection_details_from_query(
+    configuration: &configuration::Configuration,
+    query_request: &models::QueryRequest,
+) -> Result<Option<RequestConnectionDetails>, translation::error::Error> {
+    // check in config if we need to look in request arguments
+    match &configuration.connection {
+        ConnectionSettings::Static { .. } => Ok(None),
+        ConnectionSettings::Dynamic {
+            dynamic_connection_settings: DynamicConnectionSettings::NamedFromList,
+        } => {
+            let connection_identifier = &query_request
+                .request_arguments
+                .as_ref()
+                .and_then(|request_arguments| request_arguments.get("connection_identifier"))
+                .ok_or_else(|| translation::error::Error::MissingConnectionIdentifier)?;
+
+            // lookup uri
+            let connection_uri = "";
+
+            let pool = create_pool(connection_uri, environment, pool_settings)
+                .instrument(info_span!(
+                    "Create connection pool",
+                    internal.visibility = "user",
+                ))
+                .await?;
+
+            Ok(Some(RequestConnectionDetails {
+                pool 
+            }))
+        }
+    }
+}
+
 fn plan_query(
     configuration: &configuration::Configuration,
     state: &state::State,
@@ -79,15 +124,18 @@ fn plan_query(
 async fn execute_query(
     state: &state::State,
     plan: sql::execution_plan::ExecutionPlan<sql::execution_plan::Query>,
+    connection_details: Option<RequestConnectionDetails>,
 ) -> Result<JsonResponse<models::QueryResponse>, query_engine_execution::error::Error> {
-    let state::Pool::Static {
-        pool,
-        database_info,
-    } = &state.pool
-    else {
-        todo!("Dynamic connect for execute_query");
-    };
-
+    match (&state.pool, connection_details) {
+        (
+            state::Pool::Static {
+                pool,
+                database_info,
+            },
+            _,
+        ) => (&pool,Some(database_info))
+        (state::Pool::Dynamic(_), Some(RequestConnectionDetails { pool })) => (&pool,None),
+    }
     query_engine_execution::query::execute(pool, database_info, &state.query_metrics, plan)
         .await
         .map(JsonResponse::Serialized)

crates/connectors/ndc-postgres/src/state.rs

@@ -114,7 +114,7 @@ pub async fn create_state(
 
 /// Create a connection pool with default settings.
 /// - <https://docs.rs/sqlx/latest/sqlx/pool/struct.PoolOptions.html>
-async fn create_pool(
+pub async fn create_pool(
     connection_url: &str,
     environment: impl Environment,
     pool_settings: &PoolSettings,

crates/query-engine/translation/src/translation/error.rs

@@ -68,6 +68,7 @@ pub enum Error {
         tables_in_scope_names: Vec<String>,
         scope: usize,
     },
+    MissingConnectionIdentifier,
 }
 
 /// Capabilities we don't currently support.
@@ -262,6 +263,12 @@ impl std::fmt::Display for Error {
                 }
                 write!(f, "].")
             }
+            Error::MissingConnectionIdentifier => {
+                write!(
+                    f,
+                    "Expected to find connection_identifier in query_arguments but did not"
+                )
+            }
         }
     }
 }