rust: Use SQL formatting

clippy.toml

@@ -53,6 +53,20 @@ disallowed-methods = [
     { path = "aws_config::from_env", reason = "use the `mz_aws_config::defaults` function instead" },
     { path = "aws_config::load_from_env", reason = "use the `mz_aws_config::defaults` function instead" },
     { path = "aws_sdk_s3::Client::new", reason = "use the `mz_aws_util::s3::new_client` function instead" },
+    { path = "tokio_postgres::Client::simple_query", reason = "use `mz_postgres_util::simple_query` with `Sql` instead" },
+    { path = "tokio_postgres::Client::batch_execute", reason = "use `mz_postgres_util::batch_execute` with `Sql` instead" },
+    { path = "tokio_postgres::Client::query", reason = "use `mz_postgres_util::query`/`query_prepared` wrappers instead" },
+    { path = "tokio_postgres::Client::query_one", reason = "use `mz_postgres_util::query_one`/`query_one_prepared` wrappers instead" },
+    { path = "tokio_postgres::Client::query_opt", reason = "use `mz_postgres_util::query_opt`/`query_opt_prepared` wrappers instead" },
+    { path = "tokio_postgres::Client::execute", reason = "use `mz_postgres_util::execute`/`execute_prepared` wrappers instead" },
+    { path = "tokio_postgres::Transaction::simple_query", reason = "use `Sql` wrappers from `mz_postgres_util` instead" },
+    { path = "tokio_postgres::Transaction::batch_execute", reason = "use `Sql` wrappers from `mz_postgres_util` instead" },
+    { path = "tokio_postgres::Transaction::query", reason = "use `mz_postgres_util::query`/`query_prepared` wrappers instead" },
+    { path = "tokio_postgres::Transaction::query_one", reason = "use `mz_postgres_util::query_one`/`query_one_prepared` wrappers instead" },
+    { path = "tokio_postgres::Transaction::query_opt", reason = "use `mz_postgres_util::query_opt`/`query_opt_prepared` wrappers instead" },
+    { path = "tokio_postgres::Transaction::execute", reason = "use `mz_postgres_util::execute`/`execute_prepared` wrappers instead" },
+    { path = "postgres::Client::simple_query", reason = "avoid direct string SQL; use `Sql` composition and wrappers where possible" },
+    { path = "postgres::Client::batch_execute", reason = "avoid direct string SQL; use `Sql` composition and wrappers where possible" },
 
     # Prevent access to Differential APIs that want to use the default trace or use a default name, or where we offer
     # our own wrapper

src/adapter/src/catalog.rs

@@ -2328,6 +2328,7 @@ mod tests {
 
     use itertools::Itertools;
     use mz_catalog::memory::objects::CatalogItem;
+    use mz_postgres_util::{query, sql};
     use tokio_postgres::NoTls;
     use tokio_postgres::types::Type;
     use uuid::Uuid;
@@ -2877,39 +2878,43 @@ mod tests {
                 name: String,
             }
 
-            let pg_proc: BTreeMap<_, _> = client
-                .query(
+            let pg_proc: BTreeMap<_, _> = query(
+                &client,
+                sql!(
                     "SELECT
                     p.oid,
                     proname,
                     proargtypes,
                     prorettype,
                     proretset
                 FROM pg_proc p
-                JOIN pg_namespace n ON p.pronamespace = n.oid",
-                    &[],
-                )
-                .await
-                .expect("pg query failed")
-                .into_iter()
-                .map(|row| {
-                    let oid: u32 = row.get("oid");
-                    let pg_proc = PgProc {
-                        name: row.get("proname"),
-                        arg_oids: row.get("proargtypes"),
-                        ret_oid: row.get("prorettype"),
-                        ret_set: row.get("proretset"),
-                    };
-                    (oid, pg_proc)
-                })
-                .collect();
+                JOIN pg_namespace n ON p.pronamespace = n.oid"
+                ),
+                &[],
+            )
+            .await
+            .expect("pg query failed")
+            .into_iter()
+            .map(|row| {
+                let oid: u32 = row.get("oid");
+                let pg_proc = PgProc {
+                    name: row.get("proname"),
+                    arg_oids: row.get("proargtypes"),
+                    ret_oid: row.get("prorettype"),
+                    ret_set: row.get("proretset"),
+                };
+                (oid, pg_proc)
+            })
+            .collect();
 
-            let pg_type: BTreeMap<_, _> = client
-                .query(
-                    "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive FROM pg_type",
-                    &[],
-                )
-                .await
+            let pg_type: BTreeMap<_, _> = query(
+                &client,
+                sql!(
+                    "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive FROM pg_type"
+                ),
+                &[],
+            )
+            .await
                 .expect("pg query failed")
                 .into_iter()
                 .map(|row| {
@@ -2926,20 +2931,23 @@ mod tests {
                 })
                 .collect();
 
-            let pg_oper: BTreeMap<_, _> = client
-                .query("SELECT oid, oprname, oprresult FROM pg_operator", &[])
-                .await
-                .expect("pg query failed")
-                .into_iter()
-                .map(|row| {
-                    let oid: u32 = row.get("oid");
-                    let pg_oper = PgOper {
-                        name: row.get("oprname"),
-                        oprresult: row.get("oprresult"),
-                    };
-                    (oid, pg_oper)
-                })
-                .collect();
+            let pg_oper: BTreeMap<_, _> = query(
+                &client,
+                sql!("SELECT oid, oprname, oprresult FROM pg_operator"),
+                &[],
+            )
+            .await
+            .expect("pg query failed")
+            .into_iter()
+            .map(|row| {
+                let oid: u32 = row.get("oid");
+                let pg_oper = PgOper {
+                    name: row.get("oprname"),
+                    oprresult: row.get("oprresult"),
+                };
+                (oid, pg_oper)
+            })
+            .collect();
 
             let conn_catalog = catalog.for_system_session();
             let resolve_type_oid = |item: &str| {

src/balancerd/tests/server.rs

@@ -47,6 +47,7 @@ use uuid::Uuid;
 
 #[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))]
 #[cfg_attr(miri, ignore)] // too slow
+#[allow(clippy::disallowed_methods)]
 async fn test_balancer() {
     let ca = Ca::new_root("test ca").unwrap();
     let (server_cert, server_key) = ca

src/environmentd/Cargo.toml

@@ -69,6 +69,7 @@ mz-persist-client = { path = "../persist-client" }
 mz-pgrepr = { path = "../pgrepr" }
 mz-pgwire = { path = "../pgwire" }
 mz-pgwire-common = { path = "../pgwire-common" }
+mz-postgres-util = { path = "../postgres-util" }
 mz-prof-http = { path = "../prof-http" }
 mz-repr = { path = "../repr" }
 mz-secrets = { path = "../secrets" }