feat(postgres): set options through setting run-time parameters

sqlx-core/src/postgres/connection/establish.rs

@@ -40,8 +40,8 @@ impl PgConnection {
             params.push(("application_name", application_name));
         }
 
-        if let Some(ref options) = options.options {
-            params.push(("options", options));
+        for (k, v) in options.options.iter() {
+            params.push((k, v));
         }
 
         stream

sqlx-core/src/postgres/mod.rs

@@ -32,6 +32,7 @@ pub use error::{PgDatabaseError, PgErrorPosition};
 pub use listener::{PgListener, PgNotification};
 pub use message::PgSeverity;
 pub use options::{PgConnectOptions, PgSslMode};
+pub(crate) use options::parse_options;
 pub use query_result::PgQueryResult;
 pub use row::PgRow;
 pub use statement::PgStatement;

sqlx-core/src/postgres/options/mod.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::env::var;
 use std::path::{Path, PathBuf};
 
@@ -33,7 +34,7 @@ pub use ssl_mode::PgSslMode;
 /// | `password` | `None` | Password to be used if the server demands password authentication. |
 /// | `port` | `5432` | Port number to connect to at the server host, or socket file name extension for Unix-domain connections. |
 /// | `dbname` | `None` | The database name. |
-/// | `options` | `None` | The command-line options to send to the server at connection start. |
+/// | `options` | `None` | The runtime parameters to send to the server at connection start. |
 ///
 /// The URI scheme designator can be either `postgresql://` or `postgres://`.
 /// Each of the URI parts is optional.
@@ -86,7 +87,7 @@ pub struct PgConnectOptions {
     pub(crate) statement_cache_capacity: usize,
     pub(crate) application_name: Option<String>,
     pub(crate) log_settings: LogSettings,
-    pub(crate) options: Option<String>,
+    pub(crate) options: HashMap<String, String>,
 }
 
 impl Default for PgConnectOptions {
@@ -147,7 +148,9 @@ impl PgConnectOptions {
             statement_cache_capacity: 100,
             application_name: var("PGAPPNAME").ok(),
             log_settings: Default::default(),
-            options: var("PGOPTIONS").ok(),
+            options: var("PGOPTIONS")
+                .and_then(|v| Ok(parse_options(&v).unwrap_or_default().into_iter().collect()))
+                .unwrap_or_default(),
         }
     }
 
@@ -329,10 +332,17 @@ impl PgConnectOptions {
     /// ```rust
     /// # use sqlx_core::postgres::PgConnectOptions;
     /// let options = PgConnectOptions::new()
-    ///     .options("-c geqo=off -c statement_timeout=5min");
+    ///     .options(&[("geqo", "off"), ("statement_timeout", "5min")]);
     /// ```
-    pub fn options(mut self, options: &str) -> Self {
-        self.options = Some(options.to_owned());
+    pub fn options<K, V, I>(mut self, options: I) -> Self
+    where
+        K: ToString,
+        V: ToString,
+        I: IntoIterator<Item = (K, V)>,
+    {
+        for (k, v) in options {
+            self.options.insert(k.to_string(), v.to_string());
+        }
         self
     }
 
@@ -353,6 +363,24 @@ impl PgConnectOptions {
     }
 }
 
+/// Parse a libpq style options string
+pub(crate) fn parse_options(input: &str) -> Option<Vec<(String, String)>> {
+    let mut options = Vec::new();
+    for part in input.split(' ') {
+        let part = part.trim();
+        if part.is_empty() || part == "-c" {
+            continue;
+        }
+        let pair = part.splitn(2, '=').collect::<Vec<_>>();
+        if pair.len() != 2 {
+            return None;
+        }
+        options.push((pair[0].to_string(), pair[1].to_string()));
+    }
+
+    Some(options)
+}
+
 fn default_host(port: u16) -> String {
     // try to check for the existence of a unix socket and uses that
     let socket = format!(".s.PGSQL.{}", port);

sqlx-core/src/postgres/options/parse.rs

@@ -1,5 +1,5 @@
 use crate::error::Error;
-use crate::postgres::PgConnectOptions;
+use crate::postgres::{parse_options, PgConnectOptions};
 use percent_encoding::percent_decode_str;
 use std::net::IpAddr;
 use std::str::FromStr;
@@ -85,7 +85,17 @@ impl FromStr for PgConnectOptions {
 
                 "application_name" => options = options.application_name(&*value),
 
-                "options" => options = options.options(&*value),
+                "options" => {
+                    if let Some(value) = parse_options(&value) {
+                        options = options.options(value);
+                    }
+                }
+
+                k if k.starts_with("options[") => {
+                    if let Some(key) = k.strip_prefix("options[").unwrap().strip_suffix(']') {
+                        options = options.options([(key, &*value)]);
+                    }
+                }
 
                 _ => log::warn!("ignoring unrecognized connect parameter: {}={}", key, value),
             }
@@ -198,9 +208,30 @@ fn it_parses_socket_correctly_with_username_percent_encoded() {
     assert_eq!(Some("database"), opts.database.as_deref());
 }
 #[test]
-fn it_parses_options() {
-    let uri = "postgres:///?options=-c%20synchronous_commit%3Doff%20--search_path%3Dpostgres";
+fn it_parses_libpq_options_correctly() {
+    let uri = "postgres:///?options=-c%20synchronous_commit%3Doff%20-c%20search_path%3Dpostgres";
+    let opts = PgConnectOptions::from_str(uri).unwrap();
+
+    assert_eq!(
+        Some(&"off".to_string()),
+        opts.options.get("synchronous_commit")
+    );
+    assert_eq!(
+        Some(&"postgres".to_string()),
+        opts.options.get("search_path")
+    );
+}
+#[test]
+fn it_parses_sqlx_options_correctly() {
+    let uri = "postgres:///?options[synchronous_commit]=off&options[search_path]=postgres";
     let opts = PgConnectOptions::from_str(uri).unwrap();
 
-    assert_eq!(Some("-c synchronous_commit=off --search_path=postgres"), opts.options.as_deref());
+    assert_eq!(
+        Some(&"off".to_string()),
+        opts.options.get("synchronous_commit")
+    );
+    assert_eq!(
+        Some(&"postgres".to_string()),
+        opts.options.get("search_path")
+    );
 }