diff --git a/sqlx-postgres/src/connection/executor.rs b/sqlx-postgres/src/connection/executor.rs index e0f4c3d44a..a2e7ba0637 100644 --- a/sqlx-postgres/src/connection/executor.rs +++ b/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::().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,7 +79,7 @@ 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::(Some(parameters), row_desc, resolve_column_origin) .await?; @@ -87,9 +87,24 @@ async fn prepare( // 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: ¶m_types, + query: sql, + statement: id, + })?; + conn.write_sync(); + conn.inner.stream.flush().await?; + conn.inner.stream.recv_expect::().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::( + let (new_metadata, _) = self.resolve_statement_metadata::( None, Some(message.decode()?), false, diff --git a/sqlx-postgres/src/connection/resolve.rs b/sqlx-postgres/src/connection/resolve.rs index e47c318926..7ebe21465e 100644 --- a/sqlx-postgres/src/connection/resolve.rs +++ b/sqlx-postgres/src/connection/resolve.rs @@ -32,11 +32,13 @@ impl PgConnection { param_desc: Option, row_desc: Option, resolve_column_origin: bool, - ) -> Result, Error> { + ) -> Result<(Arc, 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 { @@ -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 { let mut missing_dependencies = HashMap::>::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 { 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) } } diff --git a/tests/postgres/postgres.rs b/tests/postgres/postgres.rs index 126771565a..1a19e30bc1 100644 --- a/tests/postgres/postgres.rs +++ b/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::().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::().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()