fix(postgres): re-parse unnamed statement after resolving custom-type OIDs

sqlx-postgres/src/connection/executor.rs

@@ -65,12 +65,12 @@ async fn prepare(
     // indicates that the SQL query string is now successfully parsed and has semantic validity
     conn.inner.stream.recv_expect::<ParseComplete>().await?;
 
-    let metadata = if let Some(metadata) = metadata {
+    let (metadata, ran_catalog_query) = if let Some(metadata) = metadata {
         // each SYNC produces one READY FOR QUERY
         conn.recv_ready_for_query().await?;
 
         // we already have metadata
-        metadata
+        (metadata, false)
     } else {
         let parameters = recv_desc_params(conn).await?;
 
@@ -79,17 +79,32 @@ async fn prepare(
         // each SYNC produces one READY FOR QUERY
         conn.recv_ready_for_query().await?;
 
-        let metadata = conn
+        let (metadata, ran_catalog_query) = conn
             .resolve_statement_metadata::<true>(Some(parameters), row_desc, resolve_column_origin)
             .await?;
 
         // ensure that if we did fetch custom data, we wait until we are fully ready before
         // continuing
         conn.wait_until_ready().await?;
 
-        metadata
+        (metadata, ran_catalog_query)
     };
 
+    // Resolving custom-type OIDs above goes through the simple query protocol, which
+    // destroys the unnamed prepared statement. Re-parse it so the Bind that follows can
+    // reference it; named (persistent) statements are unaffected. See #4305.
+    if !persistent && ran_catalog_query {
+        conn.inner.stream.write_msg(Parse {
+            param_types: &param_types,
+            query: sql,
+            statement: id,
+        })?;
+        conn.write_sync();
+        conn.inner.stream.flush().await?;
+        conn.inner.stream.recv_expect::<ParseComplete>().await?;
+        conn.recv_ready_for_query().await?;
+    }
+
     Ok((id, metadata))
 }
 
@@ -333,7 +348,7 @@ impl PgConnection {
 
                     // indicates that a *new* set of rows are about to be returned
                     BackendMessageFormat::RowDescription => {
-                        let new_metadata = self.resolve_statement_metadata::<false>(
+                        let (new_metadata, _) = self.resolve_statement_metadata::<false>(
                             None,
                             Some(message.decode()?),
                             false,

sqlx-postgres/src/connection/resolve.rs

@@ -32,11 +32,13 @@ impl PgConnection {
         param_desc: Option<ParameterDescription>,
         row_desc: Option<RowDescription>,
         resolve_column_origin: bool,
-    ) -> Result<Arc<PgStatementMetadata>, Error> {
+    ) -> Result<(Arc<PgStatementMetadata>, bool), Error> {
         let param_types = param_desc.map_or_else(Default::default, |desc| desc.types);
 
         let fields = row_desc.map_or_else(Default::default, |desc| desc.fields);
 
+        let mut ran_query = false;
+
         if QUERIES_ALLOWED {
             let mut type_resolver = TypeResolver::default();
             let mut column_resolver = ColumnResolver::default();
@@ -62,10 +64,12 @@ impl PgConnection {
             }
 
             // No-op if `.push_type()` was not called
-            type_resolver.fill_cache(self).await?;
+            let type_ran = type_resolver.fill_cache(self).await?;
 
             // No-op if `.push_column()` was not called
-            column_resolver.fill_cache(self).await?;
+            let column_ran = column_resolver.fill_cache(self).await?;
+
+            ran_query = type_ran || column_ran;
         }
 
         let mut parameters = Vec::with_capacity(param_types.len());
@@ -109,11 +113,14 @@ impl PgConnection {
             column_names.insert(name, ordinal);
         }
 
-        Ok(Arc::new(PgStatementMetadata {
-            columns,
-            column_names: column_names.into(),
-            parameters,
-        }))
+        Ok((
+            Arc::new(PgStatementMetadata {
+                columns,
+                column_names: column_names.into(),
+                parameters,
+            }),
+            ran_query,
+        ))
     }
 
     fn try_table_column(&self, relation_oid: Oid, attribute_no: i16) -> Option<TableColumn> {
@@ -396,8 +403,9 @@ impl TypeResolver {
         }
     }
 
-    async fn fill_cache(&mut self, conn: &mut PgConnection) -> Result<(), Error> {
+    async fn fill_cache(&mut self, conn: &mut PgConnection) -> Result<bool, Error> {
         let mut missing_dependencies = HashMap::<Oid, Vec<TypeResolverRow>>::new();
+        let mut ran_query = false;
 
         // Iteratively resolve types until all are resolved, or we hit a dead-end.
         // We statically cap the number of iterations in case we somehow encounter a circular type
@@ -406,6 +414,7 @@ impl TypeResolver {
             if self.query.is_empty() {
                 break;
             }
+            ran_query = true;
 
             // * Cancel-safety
             // * Makes this type reusable if we want to for whatever reason
@@ -476,7 +485,7 @@ impl TypeResolver {
             )));
         }
 
-        Ok(())
+        Ok(ran_query)
     }
 }
 
@@ -542,9 +551,9 @@ impl ColumnResolver {
         }
     }
 
-    async fn fill_cache(&mut self, conn: &mut PgConnection) -> Result<(), Error> {
+    async fn fill_cache(&mut self, conn: &mut PgConnection) -> Result<bool, Error> {
         if self.query.is_empty() {
-            return Ok(());
+            return Ok(false);
         }
 
         let mut query = mem::take(&mut self.query);
@@ -571,7 +580,7 @@ impl ColumnResolver {
             table_columns.columns.extend(row.columns);
         }
 
-        Ok(())
+        Ok(true)
     }
 }
 

tests/postgres/postgres.rs

@@ -1919,6 +1919,28 @@ CREATE TABLE issue_1254 (id INT4 PRIMARY KEY, pairs PAIR[]);
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn test_issue_4305() -> anyhow::Result<()> {
+    let mut setup = new::<Postgres>().await?;
+    setup
+        .execute(
+            "DROP TYPE IF EXISTS issue_4305 CASCADE; CREATE TYPE issue_4305 AS ENUM ('a', 'b');",
+        )
+        .await?;
+
+    // A fresh connection so the type-OID cache is cold. A non-persistent query
+    // returning a custom type used to fail with "unnamed prepared statement does
+    // not exist" (SQLSTATE 26000): resolving the type's OID runs a simple query
+    // that destroyed the just-parsed unnamed statement.
+    let mut conn = new::<Postgres>().await?;
+    sqlx::query("SELECT 'a'::issue_4305")
+        .persistent(false)
+        .fetch_one(&mut conn)
+        .await?;
+
+    Ok(())
+}
+
 #[sqlx_macros::test]
 async fn test_advisory_locks() -> anyhow::Result<()> {
     let pool = PgPoolOptions::new()