From f629e34063965a07b2850fc68d3f8ad1fe252d1c Mon Sep 17 00:00:00 2001 From: Dipesh Babu Date: Sun, 20 Sep 2026 14:22:02 -0600 Subject: [PATCH 1/5] fix(protocol): preserve extended SQL PREPARE replies --- .../python/test_extended_sql_prepare.py | 58 +++++++++++++++++++ pgdog/src/backend/prepared_statements.rs | 37 ++++++++++++ .../test/rewrite_simple_prepared.rs | 32 ++++++++++ .../router/parser/rewrite/statement/plan.rs | 21 ++++++- pgdog/src/net/protocol_message.rs | 19 ++++-- 5 files changed, 161 insertions(+), 6 deletions(-) create mode 100644 integration/python/test_extended_sql_prepare.py diff --git a/integration/python/test_extended_sql_prepare.py b/integration/python/test_extended_sql_prepare.py new file mode 100644 index 000000000..5ec252008 --- /dev/null +++ b/integration/python/test_extended_sql_prepare.py @@ -0,0 +1,58 @@ +import uuid + +import asyncpg +import pytest +from globals import admin, no_out_of_sync, normal_async, sharded_async + + +@pytest.fixture +def full_prepared_statements(): + with admin() as connection: + connection.execute("SET prepared_statements TO 'full'") + try: + yield + finally: + connection.execute("RELOAD") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("connect", [normal_async, sharded_async]) +@pytest.mark.parametrize("extended_prepare", [False, True]) +async def test_sql_prepare_execute_extended(connect, extended_prepare, full_prepared_statements): + connection = await connect() + name = "extended_sql_" + uuid.uuid4().hex + try: + sql = f"PREPARE {name} AS SELECT $1::integer * 2" + if extended_prepare: + outer = await connection.prepare(sql, timeout=5) + assert outer.get_parameters() == () + assert outer.get_attributes() == () + for _ in range(3): + assert await outer.fetch(timeout=5) == [] + assert outer.get_statusmsg() == "PREPARE" + else: + assert await connection.execute(sql, timeout=5) == "PREPARE" + + execute = await connection.prepare(f"EXECUTE {name}(21)", timeout=5) + for _ in range(3): + assert [tuple(row) for row in await execute.fetch(timeout=5)] == [(42,)] + assert await connection.fetchval("SELECT 1", timeout=5) == 1 + no_out_of_sync() + finally: + await connection.close(timeout=5) + + +@pytest.mark.asyncio +async def test_extended_sql_prepare_error_recovers(full_prepared_statements): + connection = await normal_async() + name = "extended_sql_error_" + uuid.uuid4().hex + try: + outer = await connection.prepare( + f"PREPARE {name} AS SELECT nonexistent_issue1403_column", timeout=5 + ) + with pytest.raises(asyncpg.UndefinedColumnError): + await outer.fetch(timeout=5) + assert await connection.fetchval("SELECT 1", timeout=5) == 1 + no_out_of_sync() + finally: + await connection.close(timeout=5) diff --git a/pgdog/src/backend/prepared_statements.rs b/pgdog/src/backend/prepared_statements.rs index 080bb0c5b..6b303c0f1 100644 --- a/pgdog/src/backend/prepared_statements.rs +++ b/pgdog/src/backend/prepared_statements.rs @@ -278,6 +278,22 @@ impl PreparedStatements { self.state.add(ExecutionCode::ExecutionCompleted); } + ProtocolMessage::ExecutePrepare { prepare, .. } => { + if self.contains(prepare.name()) { + // SQL statements use global names too, so another client may + // already have prepared this query on the pooled connection. + let reply = if self.server_state == State::TransactionError { + ErrorResponse::in_failed_transaction().message() + } else { + crate::net::CommandComplete::from_str("PREPARE").message() + }; + self.state.add_simulated(reply); + return Ok(HandleResult::Drop); + } + self.parses.push_back(prepare.name().to_owned()); + self.state.add(ExecutionCode::ExecutionCompleted); + } + ProtocolMessage::Sync(_) => { self.state.add(ExecutionCode::ReadyForQuerySync); } @@ -994,6 +1010,27 @@ pub(crate) mod test { ); } + #[test] + fn extended_sql_prepare_tracks_completion_without_ready_for_query() { + let mut ps = new_extended(); + let name = "__stmt_extended_prepare"; + let execute = ProtocolMessage::ExecutePrepare { + execute: crate::net::Execute::new(), + prepare: SimplePrepare::new(name, "PREPARE __pgdog_template_name AS SELECT $1"), + }; + + assert_eq!(ps.handle(&execute).expect("execute"), HandleResult::Forward); + assert!(!ps.contains(name)); + let mut complete = CommandComplete::from_str("PREPARE").message(); + assert!(ps.forward(&mut complete).expect("command complete")); + assert!(ps.contains(name)); + assert!(ps.done()); + assert_eq!( + ps.handle(&execute).expect("cached prepare"), + HandleResult::Drop + ); + } + #[test] fn ensure_prepared_completes_after_backend_responses() { let mut ps = new_extended(); diff --git a/pgdog/src/frontend/client/query_engine/test/rewrite_simple_prepared.rs b/pgdog/src/frontend/client/query_engine/test/rewrite_simple_prepared.rs index 6c8f5079d..95adfaed5 100644 --- a/pgdog/src/frontend/client/query_engine/test/rewrite_simple_prepared.rs +++ b/pgdog/src/frontend/client/query_engine/test/rewrite_simple_prepared.rs @@ -66,6 +66,38 @@ fn rewritten_query(messages: &[ProtocolMessage]) -> String { } } +#[tokio::test] +async fn test_rewrite_extended_prepare_preserves_protocol() { + load_test(); + change_config(|general| { + general.prepared_statements = PreparedStatementsLevel::Full; + }); + let mut client = Client::new_test(Stream::dev_null(), Parameters::default()); + let messages = run_test( + &mut client, + &[ + Parse::named("outer", "PREPARE inner_stmt AS SELECT 1").into(), + Bind::new_statement("outer").into(), + Describe::new_portal("").into(), + Execute::new().into(), + Sync::new().into(), + ], + ) + .await; + + assert_eq!( + messages.iter().map(Protocol::code).collect::(), + "PBDES" + ); + assert!( + matches!(&messages[0], ProtocolMessage::Parse(parse) if parse.query().starts_with("PREPARE __pgdog_")) + ); + assert!(matches!( + &messages[3], + ProtocolMessage::ExecutePrepare { .. } + )); +} + #[tokio::test] async fn test_reprepare_releases_previous_statement() { load_test(); diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs index 116b2d3eb..1a82be9fc 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs @@ -216,8 +216,25 @@ impl RewritePlan { .iter() .for_each(|prepare| match prepare { PrepareExecute::Prepare(prepare) => { - request.messages.clear(); - request.push(ProtocolMessage::PrepareFromClient(prepare.clone())); + if request + .messages + .iter() + .any(|message| matches!(message, ProtocolMessage::Query(_))) + { + request.messages.clear(); + request.push(ProtocolMessage::PrepareFromClient(prepare.clone())); + } else { + // Keep Parse/Bind/Describe/Sync and their corresponding replies. + // Preparing the outer statement must not execute the SQL PREPARE. + for message in &mut request.messages { + if let ProtocolMessage::Execute(execute) = message { + *message = ProtocolMessage::ExecutePrepare { + execute: execute.clone(), + prepare: prepare.clone(), + }; + } + } + } } PrepareExecute::Execute(prepare) => { request diff --git a/pgdog/src/net/protocol_message.rs b/pgdog/src/net/protocol_message.rs index 12b32c0d2..1fe2f07ad 100644 --- a/pgdog/src/net/protocol_message.rs +++ b/pgdog/src/net/protocol_message.rs @@ -15,6 +15,11 @@ pub(crate) enum ProtocolMessage { Describe(Describe), EnsurePrepared(Prepare), PrepareFromClient(Prepare), + /// Execute a SQL PREPARE sent through the extended protocol. + ExecutePrepare { + execute: Execute, + prepare: Prepare, + }, Execute(Execute), Close(Close), Query(Query), @@ -31,7 +36,13 @@ impl ProtocolMessage { use ProtocolMessage::*; matches!( self, - Bind(_) | Parse(_) | Describe(_) | Execute(_) | Sync(_) | Close(_) + Bind(_) + | Parse(_) + | Describe(_) + | Execute(_) + | ExecutePrepare { .. } + | Sync(_) + | Close(_) ) } @@ -64,7 +75,7 @@ impl ProtocolMessage { Self::Describe(describe) => describe.len(), Self::EnsurePrepared(prepare) => prepare.len(), Self::PrepareFromClient(prepare) => prepare.len(), - Self::Execute(execute) => execute.len(), + Self::Execute(execute) | Self::ExecutePrepare { execute, .. } => execute.len(), Self::Close(close) => close.len(), Self::Query(query) => query.len(), Self::Other(message) => message.len(), @@ -84,7 +95,7 @@ impl Protocol for ProtocolMessage { Self::Parse(parse) => parse.code(), Self::Describe(describe) => describe.code(), Self::EnsurePrepared { .. } | Self::PrepareFromClient { .. } => 'Q', - Self::Execute(execute) => execute.code(), + Self::Execute(execute) | Self::ExecutePrepare { execute, .. } => execute.code(), Self::Close(close) => close.code(), Self::Query(query) => query.code(), Self::Other(message) => message.code(), @@ -125,7 +136,7 @@ impl ToBytes for ProtocolMessage { Self::Describe(describe) => describe.to_bytes(), Self::EnsurePrepared(prepare) => prepare.to_bytes(), Self::PrepareFromClient(prepare) => prepare.to_bytes(), - Self::Execute(execute) => execute.to_bytes(), + Self::Execute(execute) | Self::ExecutePrepare { execute, .. } => execute.to_bytes(), Self::Close(close) => close.to_bytes(), Self::Query(query) => query.to_bytes(), Self::Other(message) => message.to_bytes(), From 5c8410bfd6438b8366b7ec29b314e831e3a4387f Mon Sep 17 00:00:00 2001 From: Dipesh Babu Date: Sun, 20 Sep 2026 23:48:14 -0600 Subject: [PATCH 2/5] fix(protocol): refresh SQL EXECUTE rewrites per request --- integration/elixir/test/prepared_test.exs | 21 +++++ .../python/test_extended_sql_prepare.py | 78 +++++++++++++++++++ .../ruby/prepared_disabled/prepared_spec.rb | 3 + .../ruby/prepared_extended/prepared_spec.rb | 3 + .../ruby/prepared_full/prepared_spec.rb | 3 + integration/ruby/sql_prepare_examples.rb | 29 +++++++ pgdog/src/backend/prepared_statements.rs | 68 +++++++++++++--- .../src/frontend/client/query_engine/query.rs | 2 +- pgdog/src/frontend/client_request.rs | 8 +- .../router/parser/cache/cache_impl.rs | 6 +- .../router/parser/rewrite/statement/offset.rs | 13 ++-- .../router/parser/rewrite/statement/plan.rs | 36 ++++++++- pgdog/src/net/protocol_message.rs | 21 ++++- 13 files changed, 262 insertions(+), 29 deletions(-) create mode 100644 integration/ruby/sql_prepare_examples.rb diff --git a/integration/elixir/test/prepared_test.exs b/integration/elixir/test/prepared_test.exs index 68ad3e911..437e7b9b3 100644 --- a/integration/elixir/test/prepared_test.exs +++ b/integration/elixir/test/prepared_test.exs @@ -1,6 +1,27 @@ defmodule Pgdog.PreparedTest do use ExUnit.Case, async: false + test "SQL PREPARE and EXECUTE preserve the extended protocol" do + for options <- [[], [prepare: :unnamed]] do + conn = Pgdog.connect(options) + name = "elixir_sql_#{System.unique_integer([:positive])}" + + try do + prepare = Postgrex.prepare!(conn, "prepare_command", "PREPARE #{name} AS SELECT $1::bigint * 2") + assert %Postgrex.Result{command: :prepare} = Postgrex.execute!(conn, prepare, []) + execute = Postgrex.prepare!(conn, "execute_command", "EXECUTE #{name}(21)") + + for _ <- 1..3 do + assert Pgdog.one(Postgrex.execute!(conn, execute, [])) == 42 + end + + assert Pgdog.one(Postgrex.query!(conn, "SELECT 1", [])) == 1 + after + GenServer.stop(conn) + end + end + end + test "a named statement is reusable across many executions" do conn = Pgdog.connect() query = Postgrex.prepare!(conn, "elixir_echo", "SELECT $1::bigint") diff --git a/integration/python/test_extended_sql_prepare.py b/integration/python/test_extended_sql_prepare.py index 5ec252008..e6e52194a 100644 --- a/integration/python/test_extended_sql_prepare.py +++ b/integration/python/test_extended_sql_prepare.py @@ -15,6 +15,13 @@ def full_prepared_statements(): connection.execute("RELOAD") +@pytest.fixture +def rewritten_prepared_statements(full_prepared_statements): + with admin() as connection: + connection.execute("SET rewrite_enabled TO true") + yield + + @pytest.mark.asyncio @pytest.mark.parametrize("connect", [normal_async, sharded_async]) @pytest.mark.parametrize("extended_prepare", [False, True]) @@ -56,3 +63,74 @@ async def test_extended_sql_prepare_error_recovers(full_prepared_statements): no_out_of_sync() finally: await connection.close(timeout=5) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("extended_prepare", [False, True]) +async def test_extended_execute_limit_offset(extended_prepare, rewritten_prepared_statements): + connection = await sharded_async() + schema = "extended_limit_" + uuid.uuid4().hex + try: + await connection.execute(f'CREATE SCHEMA "{schema}"') + await connection.execute(f'CREATE TABLE "{schema}".sharded (id BIGINT PRIMARY KEY)') + for value in range(1, 56): + await connection.execute(f'INSERT INTO "{schema}".sharded VALUES ($1)', value) + cases = [ + ("", "LIMIT 5 OFFSET $1", "(10)", list(range(45, 40, -1))), + ("", "LIMIT $2 OFFSET $1", "(5, 10)", list(range(50, 40, -1))), + ("", "LIMIT $1 OFFSET $2", "(5, 10)", list(range(45, 40, -1))), + ("", "LIMIT 10 OFFSET 5", "", list(range(50, 40, -1))), + ("WHERE id < $2", "LIMIT $3 OFFSET $1", "(5, 25, 10)", list(range(19, 9, -1))), + ("WHERE id = 35", "LIMIT 1 OFFSET 0", "", [35]), + ] + for index, (predicate, clause, arguments, expected) in enumerate(cases): + name = f"{schema}_{index}" + sql = f'PREPARE {name} AS SELECT id FROM "{schema}".sharded {predicate} ORDER BY id DESC {clause}' + if extended_prepare: + prepare = await connection.prepare(sql, timeout=5) + await prepare.fetch(timeout=5) + else: + await connection.execute(sql, timeout=5) + execute = await connection.prepare(f"EXECUTE {name}{arguments}", timeout=5) + for _ in range(3): + assert [row[0] for row in await execute.fetch(timeout=5)] == expected + no_out_of_sync() + finally: + await connection.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + await connection.close(timeout=5) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("extended_prepare", [False, True]) +async def test_extended_execute_generated_values(extended_prepare, rewritten_prepared_statements): + connection = await sharded_async() + name = "extended_id_" + uuid.uuid4().hex + try: + sql = f"PREPARE {name} AS SELECT pgdog.unique_id()" + if extended_prepare: + prepare = await connection.prepare(sql, timeout=5) + await prepare.fetch(timeout=5) + else: + await connection.execute(sql, timeout=5) + execute = await connection.prepare(f"EXECUTE {name}", timeout=5) + values = [await execute.fetchval(timeout=5) for _ in range(10)] + assert len(set(values)) == len(values), "each execution must generate a fresh ID" + assert all(isinstance(value, int) for value in values) + no_out_of_sync() + finally: + await connection.close(timeout=5) + + +@pytest.mark.asyncio +async def test_sql_prepare_registers_each_clients_name(full_prepared_statements): + connections = [await normal_async(), await normal_async()] + name = "shared_sql_" + uuid.uuid4().hex + try: + for value, connection in enumerate(connections): + prepare = await connection.prepare(f"PREPARE {name} AS SELECT $1::integer", timeout=5) + await prepare.fetch(timeout=5) + execute = await connection.prepare(f"EXECUTE {name}({value})", timeout=5) + assert await execute.fetchval(timeout=5) == value + finally: + for connection in connections: + await connection.close(timeout=5) diff --git a/integration/ruby/prepared_disabled/prepared_spec.rb b/integration/ruby/prepared_disabled/prepared_spec.rb index d4cc7123e..3495c51c4 100644 --- a/integration/ruby/prepared_disabled/prepared_spec.rb +++ b/integration/ruby/prepared_disabled/prepared_spec.rb @@ -1,12 +1,15 @@ # frozen_string_literal: true require_relative '../rspec_helper' +require_relative '../sql_prepare_examples' # With prepared_statements = "disabled" pgdog forwards protocol messages as-is # without rewriting or caching. describe 'prepared_statements = disabled' do after { ensure_done } + it_behaves_like 'SQL PREPARE over extended protocol', 'pgdog_session' + # Anonymous statements (empty name) are a single Parse+Bind+Execute+Sync # cycle on one backend — no state needs to survive across cycles. it 'executes anonymous parameterized queries' do diff --git a/integration/ruby/prepared_extended/prepared_spec.rb b/integration/ruby/prepared_extended/prepared_spec.rb index 0f04c1c56..ecef70ecc 100644 --- a/integration/ruby/prepared_extended/prepared_spec.rb +++ b/integration/ruby/prepared_extended/prepared_spec.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require_relative '../rspec_helper' +require_relative '../sql_prepare_examples' # Uses the main integration pgdog.toml which sets prepared_statements = "extended". # "extended" rewrites and replays named extended-protocol statements (Parse/Bind) @@ -10,6 +11,8 @@ describe 'prepared_statements = extended' do after { ensure_done } + it_behaves_like 'SQL PREPARE over extended protocol', 'pgdog_session' + # Anonymous statements (empty name) are a single Parse+Bind+Execute+Sync # cycle on one backend — no state needs to survive across cycles. it 'executes anonymous parameterized queries' do diff --git a/integration/ruby/prepared_full/prepared_spec.rb b/integration/ruby/prepared_full/prepared_spec.rb index 0a23f7a82..32cfd0fe0 100644 --- a/integration/ruby/prepared_full/prepared_spec.rb +++ b/integration/ruby/prepared_full/prepared_spec.rb @@ -1,10 +1,13 @@ # frozen_string_literal: true require_relative '../rspec_helper' +require_relative '../sql_prepare_examples' describe 'prepared_statements = full' do after { ensure_done } + it_behaves_like 'SQL PREPARE over extended protocol', 'pgdog' + # Mirror of disabled suite: anonymous statements carry no per-backend state, # so they work identically regardless of the prepared_statements setting. it 'executes anonymous parameterized queries' do diff --git a/integration/ruby/sql_prepare_examples.rb b/integration/ruby/sql_prepare_examples.rb new file mode 100644 index 000000000..99a6e40c6 --- /dev/null +++ b/integration/ruby/sql_prepare_examples.rb @@ -0,0 +1,29 @@ +# frozen_string_literal: true + +shared_examples 'SQL PREPARE over extended protocol' do |user| + it 'executes SQL PREPARE and EXECUTE through named statements' do + conn = connect('pgdog', user) + name = "sql_extended_#{SecureRandom.hex(6)}" + conn.prepare('prepare_command', "PREPARE #{name} AS SELECT $1::bigint * 2 AS val") + expect(conn.exec_prepared('prepare_command').cmd_status).to eq('PREPARE') + conn.prepare('execute_command', "EXECUTE #{name}(21)") + 3.times do + expect(conn.exec_prepared('execute_command')[0]['val'].to_i).to eq(42) + end + expect(conn.exec('SELECT 1')[0].values).to eq(['1']) + ensure + conn.close if conn && !conn.finished? + end + + it 'executes SQL PREPARE and EXECUTE through unnamed statements' do + conn = connect('pgdog', user) + name = "sql_unnamed_#{SecureRandom.hex(6)}" + expect(conn.exec_params("PREPARE #{name} AS SELECT $1::bigint * 2 AS val", []).cmd_status).to eq('PREPARE') + 3.times do + expect(conn.exec_params("EXECUTE #{name}(21)", [])[0]['val'].to_i).to eq(42) + end + expect(conn.exec('SELECT 1')[0].values).to eq(['1']) + ensure + conn.close if conn && !conn.finished? + end +end diff --git a/pgdog/src/backend/prepared_statements.rs b/pgdog/src/backend/prepared_statements.rs index 6b303c0f1..b545da775 100644 --- a/pgdog/src/backend/prepared_statements.rs +++ b/pgdog/src/backend/prepared_statements.rs @@ -101,7 +101,9 @@ pub(crate) struct PreparedStatements { local_cache: LruCache, state: ProtocolState, // Prepared statements being prepared now on the connection. - parses: VecDeque, + // Anonymous parses occupy a slot too, so their replies cannot complete a + // later named preparation in the same request. + parses: VecDeque>, // Describes being executed now on the connection. describes: VecDeque, // Statement names of every statement Describe sent (to match each ParameterDescription to its statement) @@ -169,6 +171,7 @@ impl PreparedStatements { match request { ProtocolMessage::Parse(_) => { self.state.add_ignore('1'); + self.parses.push_back(None); Ok(()) } _ => Err(Error::UnsupportedHandleIgnore(request.code())), @@ -178,6 +181,18 @@ impl PreparedStatements { /// Handle extended protocol message. pub(super) fn handle(&mut self, request: &ProtocolMessage) -> Result { match request { + ProtocolMessage::EnsureParsed(parse) => { + debug_assert!(parse.anonymous()); + self.state.add_ignore('1'); + self.parses.push_back(None); + let mut parse = parse.clone(); + if self.rewrite_parse_data_types(&mut parse) { + return Ok(HandleResult::Rewrite(ProtocolMessage::EnsureParsed(parse))); + } + } + ProtocolMessage::BindAnonymous(_) => { + self.state.add('2'); + } ProtocolMessage::Bind(bind) => { if !bind.anonymous() { let message = self.check_prepared(bind.statement())?; @@ -187,7 +202,7 @@ impl PreparedStatements { self.state.add_ignore('3'); } self.state.add_ignore('1'); - self.parses.push_back(bind.statement().to_string()); + self.parses.push_back(Some(bind.statement().to_string())); self.state.add('2'); if self.config.level.rewrite_anonymous() { message.anonymize(); @@ -230,7 +245,8 @@ impl PreparedStatements { self.state.add_ignore('3'); } self.state.add_ignore('1'); - self.parses.push_back(describe.statement().to_string()); + self.parses + .push_back(Some(describe.statement().to_string())); self.state.add(ExecutionCode::DescriptionOrNothing); // t self.state.add(ExecutionCode::DescriptionOrNothing); // T @@ -290,7 +306,7 @@ impl PreparedStatements { self.state.add_simulated(reply); return Ok(HandleResult::Drop); } - self.parses.push_back(prepare.name().to_owned()); + self.parses.push_back(Some(prepare.name().to_owned())); self.state.add(ExecutionCode::ExecutionCompleted); } @@ -313,7 +329,7 @@ impl PreparedStatements { self.state.add_simulated(ParseComplete.message()); return Ok(HandleResult::Drop); } else { - self.parses.push_back(parse.name().to_string()); + self.parses.push_back(Some(parse.name().to_string())); } // The client is sending named prepared statements, // but we're in ExtendedAnonymous mode so we rewrite @@ -322,6 +338,8 @@ impl PreparedStatements { parse.anonymize(); rewritten = true; } + } else { + self.parses.push_back(None); } self.state.add('1'); @@ -363,7 +381,7 @@ impl PreparedStatements { ); return Ok(HandleResult::Drop); } else { - self.parses.push_back(prepare.name().to_owned()); + self.parses.push_back(Some(prepare.name().to_owned())); self.state.add(ExecutionCode::ReadyForQuery); } } @@ -381,7 +399,7 @@ impl PreparedStatements { self.state.add_ignore(ExecutionCode::CommandComplete); // (the Prepare) self.state.add_ignore(ExecutionCode::ReadyForQuery); - self.parses.push_back(name.to_owned()); + self.parses.push_back(Some(name.to_owned())); // This will do Close => Prepare return Ok(HandleResult::PrependProtocolMessage( @@ -391,7 +409,7 @@ impl PreparedStatements { return Ok(HandleResult::Drop); } } else { - self.parses.push_back(prepare.name().to_string()); + self.parses.push_back(Some(prepare.name().to_string())); self.state.add_ignore('C'); // Prepare turns into a Simple Query ('Q') so it expects a regular RFQ back. @@ -454,7 +472,7 @@ impl PreparedStatements { } '1' | 'C' => { - if let Some(name) = self.parses.pop_front() { + if let Some(Some(name)) = self.parses.pop_front() { self.prepared(&name); } } @@ -465,9 +483,9 @@ impl PreparedStatements { '3' if matches!(action, Action::Ignore) => { // ok, pop_front -> push_front just to avoid borrowing issues // and not to copy the name just to remove by name - if let Some(name) = self.parses.pop_front() { + if let Some(Some(name)) = self.parses.pop_front() { self.remove(&name); - self.parses.push_front(name); + self.parses.push_front(Some(name)); } } @@ -525,7 +543,7 @@ impl PreparedStatements { /// to run something before actual client's requests fn check_prepared(&mut self, name: &str) -> Result, Error> { // Ignore if we already have a Parse in progress. - if self.parses.iter().any(|s| s == name) { + if self.parses.iter().any(|s| s.as_deref() == Some(name)) { return Ok(None); } @@ -1010,6 +1028,32 @@ pub(crate) mod test { ); } + #[test] + fn internal_parse_does_not_complete_a_later_named_parse() { + let mut ps = new_extended(); + let internal = ProtocolMessage::EnsureParsed(Parse::named("", "SELECT 1")); + let named = ProtocolMessage::Parse(Parse::named("later_named", "SELECT 2")); + assert_eq!( + ps.handle(&internal).expect("internal parse"), + HandleResult::Forward + ); + assert_eq!( + ps.handle(&named).expect("named parse"), + HandleResult::Forward + ); + assert!( + !ps.forward(&mut ParseComplete.message()) + .expect("internal reply") + ); + assert!(!ps.contains("later_named")); + assert!( + ps.forward(&mut ParseComplete.message()) + .expect("named reply") + ); + assert!(ps.contains("later_named")); + assert!(ps.done()); + } + #[test] fn extended_sql_prepare_tracks_completion_without_ready_for_query() { let mut ps = new_extended(); diff --git a/pgdog/src/frontend/client/query_engine/query.rs b/pgdog/src/frontend/client/query_engine/query.rs index 0e23a1e34..b3256450f 100644 --- a/pgdog/src/frontend/client/query_engine/query.rs +++ b/pgdog/src/frontend/client/query_engine/query.rs @@ -60,7 +60,7 @@ impl QueryEngine { // Set response format. for msg in context.client_request.messages.iter() { - if let ProtocolMessage::Bind(bind) = msg { + if let ProtocolMessage::Bind(bind) | ProtocolMessage::BindAnonymous(bind) = msg { self.backend.bind(bind)? } } diff --git a/pgdog/src/frontend/client_request.rs b/pgdog/src/frontend/client_request.rs index 9be8088a8..bec477928 100644 --- a/pgdog/src/frontend/client_request.rs +++ b/pgdog/src/frontend/client_request.rs @@ -133,10 +133,10 @@ impl ClientRequest { ProtocolMessage::Query(query) => { return Ok(Some(BufferedQuery::Query(query.clone()))); } - ProtocolMessage::Parse(parse) => { + ProtocolMessage::Parse(parse) | ProtocolMessage::EnsureParsed(parse) => { return Ok(Some(BufferedQuery::Prepared(parse.clone()))); } - ProtocolMessage::Bind(bind) => { + ProtocolMessage::Bind(bind) | ProtocolMessage::BindAnonymous(bind) => { if !bind.anonymous() { return Ok(PreparedStatements::global() .read() @@ -188,7 +188,7 @@ impl ClientRequest { /// If this buffer contains bound parameters, retrieve them. pub(crate) fn parameters(&self) -> Result, Error> { for message in &self.messages { - if let ProtocolMessage::Bind(bind) = message { + if let ProtocolMessage::Bind(bind) | ProtocolMessage::BindAnonymous(bind) = message { return Ok(Some(bind)); } } @@ -278,7 +278,7 @@ impl ClientRequest { let mut references_anonymous = false; for message in &self.messages { match message { - ProtocolMessage::Parse(_) => return false, + ProtocolMessage::Parse(_) | ProtocolMessage::EnsureParsed(_) => return false, ProtocolMessage::Bind(bind) => { if !bind.anonymous() { return false; diff --git a/pgdog/src/frontend/router/parser/cache/cache_impl.rs b/pgdog/src/frontend/router/parser/cache/cache_impl.rs index 365ba0308..65beec7ef 100644 --- a/pgdog/src/frontend/router/parser/cache/cache_impl.rs +++ b/pgdog/src/frontend/router/parser/cache/cache_impl.rs @@ -147,7 +147,11 @@ impl Cache { // subsequent uncommented lookup would hit this entry and receive an // already-rewritten plan that was built against the commented // (direct-shard) variant. - let cacheable = entry.comment_shard.is_none() || entry.rewrite_plan.is_empty(); + // SQL PREPARE registers a client-local name. SQL EXECUTE resolves that + // name and may materialize IDs or timestamps for this execution. Neither + // rewritten plan can be reused across requests or client connections. + let cacheable = entry.rewrite_plan.prepare_rewrites.is_empty() + && (entry.comment_shard.is_none() || entry.rewrite_plan.is_empty()); if cacheable { guard .queries diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs b/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs index 166885593..380ac24c2 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs @@ -168,12 +168,15 @@ impl OffsetPlan { let new_execute = new_execute.deref(); let new_execute_sql = deparse(new_execute)?; - // `ExecuteStmt` will be a Query, because this is simple-protocol. - // Replace with our re-written `ExecuteStmt` - // (replacing limit/offset with proper multi-shard vlaues) + // SQL EXECUTE can also arrive through the extended protocol. Update its + // internal Parse so cached outer statements receive the current limits. for message in request.messages.iter_mut() { - if let ProtocolMessage::Query(query) = message { - query.set_query(new_execute_sql.as_str()); + match message { + ProtocolMessage::Query(query) => query.set_query(new_execute_sql.as_str()), + ProtocolMessage::Parse(parse) | ProtocolMessage::EnsureParsed(parse) => { + parse.set_query(new_execute_sql.as_str()); + } + _ => {} } } diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs index 1a82be9fc..1bbddf435 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs @@ -7,7 +7,7 @@ use super::{ }; use crate::frontend::client::QueryTimestamps; use crate::frontend::router::parser::rewrite::statement::non_deterministic_funcs::NDFunction; -use crate::frontend::{ClientRequest, PreparedStatements}; +use crate::frontend::{BufferedQuery, ClientRequest, PreparedStatements}; use crate::net::messages::bind::{Format, Parameter}; use crate::net::{Bind, Parse, ProtocolMessage, Query, parameter::ParameterValue}; use crate::unique_id::UniqueId; @@ -252,7 +252,11 @@ impl RewritePlan { anonymous_client_params = self.apply_parse(parse); } ProtocolMessage::Query(query) => self.apply_query(query).await?, - ProtocolMessage::Bind(bind) => self.apply_bind(bind, timezone, timestamps).await?, + ProtocolMessage::Bind(bind) if self.prepare_rewrites.is_empty() => { + // SQL PREPARE's placeholders belong to the inner statement. + // SQL EXECUTE already has its generated values in the SQL text. + self.apply_bind(bind, timezone, timestamps).await? + } _ => {} } } @@ -266,6 +270,34 @@ impl RewritePlan { request.anonymous_client_params = anonymous_client_params; + if self + .prepare_rewrites + .iter() + .any(|rewrite| matches!(rewrite, PrepareExecute::Execute(_))) + && let Some(bind_index) = request + .messages + .iter() + .position(|message| matches!(message, ProtocolMessage::Bind(_))) + && let Some(BufferedQuery::Prepared(mut parse)) = request.query()? + { + // A cached outer EXECUTE can contain per-execution values and shard- + // dependent LIMIT/OFFSET. Parse it again without adding cache entries. + // Keep the original Bind name for the cross-shard result decoder. + parse.anonymize(); + request.anonymous_client_params = self.apply_parse(&mut parse); + for message in &mut request.messages { + if let ProtocolMessage::Bind(bind) = message { + *message = ProtocolMessage::BindAnonymous(bind.clone()); + } + } + request.last_parse = None; + // EnsurePrepared is a simple Query and destroys unnamed statements, + // so the internal Parse must follow it and precede Bind. + request + .messages + .insert(bind_index, ProtocolMessage::EnsureParsed(parse)); + } + self.apply_after_messages(request) } diff --git a/pgdog/src/net/protocol_message.rs b/pgdog/src/net/protocol_message.rs index 1fe2f07ad..3371557e3 100644 --- a/pgdog/src/net/protocol_message.rs +++ b/pgdog/src/net/protocol_message.rs @@ -11,7 +11,11 @@ use super::{ #[derive(Debug, Clone, PartialEq)] pub(crate) enum ProtocolMessage { Bind(Bind), + /// Bind the unnamed rewrite while retaining the original statement's result metadata. + BindAnonymous(Bind), Parse(Parse), + /// Internal anonymous Parse whose ParseComplete is consumed by PgDog. + EnsureParsed(Parse), Describe(Describe), EnsurePrepared(Prepare), PrepareFromClient(Prepare), @@ -37,7 +41,9 @@ impl ProtocolMessage { matches!( self, Bind(_) + | BindAnonymous(_) | Parse(_) + | EnsureParsed(_) | Describe(_) | Execute(_) | ExecutePrepare { .. } @@ -51,6 +57,7 @@ impl ProtocolMessage { match self { Bind(bind) => bind.anonymous(), + BindAnonymous(_) | EnsureParsed(_) => true, Parse(parse) => parse.anonymous(), Describe(describe) => describe.anonymous(), _ => false, @@ -71,7 +78,8 @@ impl ProtocolMessage { pub(crate) fn len(&self) -> usize { match self { Self::Bind(bind) => bind.len(), - Self::Parse(parse) => parse.len(), + Self::BindAnonymous(bind) => bind.len() - bind.statement().len(), + Self::Parse(parse) | Self::EnsureParsed(parse) => parse.len(), Self::Describe(describe) => describe.len(), Self::EnsurePrepared(prepare) => prepare.len(), Self::PrepareFromClient(prepare) => prepare.len(), @@ -91,8 +99,8 @@ impl ProtocolMessage { impl Protocol for ProtocolMessage { fn code(&self) -> char { match self { - Self::Bind(bind) => bind.code(), - Self::Parse(parse) => parse.code(), + Self::Bind(bind) | Self::BindAnonymous(bind) => bind.code(), + Self::Parse(parse) | Self::EnsureParsed(parse) => parse.code(), Self::Describe(describe) => describe.code(), Self::EnsurePrepared { .. } | Self::PrepareFromClient { .. } => 'Q', Self::Execute(execute) | Self::ExecutePrepare { execute, .. } => execute.code(), @@ -132,7 +140,12 @@ impl ToBytes for ProtocolMessage { fn to_bytes(&self) -> bytes::Bytes { match self { Self::Bind(bind) => bind.to_bytes(), - Self::Parse(parse) => parse.to_bytes(), + Self::BindAnonymous(bind) => { + let mut bind = bind.clone(); + bind.anonymize(); + bind.to_bytes() + } + Self::Parse(parse) | Self::EnsureParsed(parse) => parse.to_bytes(), Self::Describe(describe) => describe.to_bytes(), Self::EnsurePrepared(prepare) => prepare.to_bytes(), Self::PrepareFromClient(prepare) => prepare.to_bytes(), From d1bf87367d22996cabcb56d7d0e7133ed890819f Mon Sep 17 00:00:00 2001 From: Dipesh Babu Date: Mon, 21 Sep 2026 00:39:56 -0600 Subject: [PATCH 3/5] fix(protocol): retain unnamed statements across executions --- .../python/test_extended_sql_prepare.py | 13 ++++++++++++ integration/ruby/sql_prepare_examples.rb | 4 ++-- pgdog/src/frontend/client/mod.rs | 20 ++++++++++++++++++- pgdog/src/frontend/client_request.rs | 2 ++ 4 files changed, 36 insertions(+), 3 deletions(-) diff --git a/integration/python/test_extended_sql_prepare.py b/integration/python/test_extended_sql_prepare.py index e6e52194a..dda1a8a76 100644 --- a/integration/python/test_extended_sql_prepare.py +++ b/integration/python/test_extended_sql_prepare.py @@ -134,3 +134,16 @@ async def test_sql_prepare_registers_each_clients_name(full_prepared_statements) finally: for connection in connections: await connection.close(timeout=5) + + +@pytest.mark.asyncio +async def test_repeated_unnamed_sql_execute(full_prepared_statements): + connection = await normal_async() + name = "unnamed_sql_" + uuid.uuid4().hex + try: + await connection.execute(f"PREPARE {name} AS SELECT 42::integer") + execute = await connection.prepare(f"EXECUTE {name}", name="", timeout=5) + for _ in range(3): + assert await execute.fetchval(timeout=5) == 42 + finally: + await connection.close(timeout=5) diff --git a/integration/ruby/sql_prepare_examples.rb b/integration/ruby/sql_prepare_examples.rb index 99a6e40c6..b210e96cd 100644 --- a/integration/ruby/sql_prepare_examples.rb +++ b/integration/ruby/sql_prepare_examples.rb @@ -5,10 +5,10 @@ conn = connect('pgdog', user) name = "sql_extended_#{SecureRandom.hex(6)}" conn.prepare('prepare_command', "PREPARE #{name} AS SELECT $1::bigint * 2 AS val") - expect(conn.exec_prepared('prepare_command').cmd_status).to eq('PREPARE') + expect(conn.exec_prepared('prepare_command', []).cmd_status).to eq('PREPARE') conn.prepare('execute_command', "EXECUTE #{name}(21)") 3.times do - expect(conn.exec_prepared('execute_command')[0]['val'].to_i).to eq(42) + expect(conn.exec_prepared('execute_command', [])[0]['val'].to_i).to eq(42) end expect(conn.exec('SELECT 1')[0].values).to eq(['1']) ensure diff --git a/pgdog/src/frontend/client/mod.rs b/pgdog/src/frontend/client/mod.rs index 96eb23e9a..0e39b1e42 100644 --- a/pgdog/src/frontend/client/mod.rs +++ b/pgdog/src/frontend/client/mod.rs @@ -30,7 +30,7 @@ use crate::net::messages::{ Authentication, BackendKeyData, ErrorResponse, FromBytes, FrontendPid, Message, Password, Protocol, ProtocolVersion, ReadyForQuery, ToBytes, scram_challenge, }; -use crate::net::{MessageBuffer, ProtocolMessage, Stream, parameter::Parameters}; +use crate::net::{MessageBuffer, Parse, ProtocolMessage, Stream, parameter::Parameters}; use crate::state::State; use crate::stats::memory::MemoryUsage; use crate::util::{safe_timeout, user_database_from_params}; @@ -85,6 +85,9 @@ pub(crate) struct Client { // This can be a query or just a `Parse` and `Flush`, but in either case, the client // will expect a response immediately and we need to handle it. client_request: ClientRequest, + // Keep the client's original unnamed statement across executions. The + // request's copy can be rewritten for whichever backend receives it. + unnamed_parse: Option, // Raw buffer of messages the client sent. We keep them here to avoid memory allocations // down the line (using [`bytes::Bytes`]). stream_buffer: MessageBuffer, @@ -430,6 +433,7 @@ impl Client { transaction: None, timeouts: Timeouts::from_config(&config.config.general), client_request: ClientRequest::default(), + unnamed_parse: None, stream_buffer: MessageBuffer::new( config.config.memory.message_buffer, config.config.general.frontend_query_size_limit_block(), @@ -469,6 +473,7 @@ impl Client { transaction: None, timeouts: Timeouts::from_config(&config().config.general), client_request: ClientRequest::default(), + unnamed_parse: None, stream_buffer: MessageBuffer::new( 4096, config().config.general.frontend_query_size_limit_block(), @@ -652,6 +657,7 @@ impl Client { cancellation_token: &CancellationToken, ) -> Result { self.client_request.clear(); + self.client_request.last_parse = self.unnamed_parse.clone(); // Check config once per request. let config = config::config(); @@ -709,6 +715,16 @@ impl Client { return Ok(BufferEvent::DisconnectGraceful); } else { let message = ProtocolMessage::from_bytes(message.to_bytes())?; + match &message { + ProtocolMessage::Parse(parse) if parse.anonymous() => { + self.unnamed_parse = Some(parse.clone()); + } + ProtocolMessage::Query(_) => self.unnamed_parse = None, + ProtocolMessage::Close(close) if close.is_statement() && close.anonymous() => { + self.unnamed_parse = None; + } + _ => {} + } self.client_request.push(message); } } @@ -779,6 +795,8 @@ impl MemoryUsage for Client { + std::mem::size_of::() + self.stream_buffer.capacity() + self.client_request.memory_usage() + + std::mem::size_of::>() + + self.unnamed_parse.as_ref().map_or(0, Parse::len) } } diff --git a/pgdog/src/frontend/client_request.rs b/pgdog/src/frontend/client_request.rs index bec477928..30b96bbf9 100644 --- a/pgdog/src/frontend/client_request.rs +++ b/pgdog/src/frontend/client_request.rs @@ -82,6 +82,8 @@ impl ClientRequest { /// Remove any saved state from the request. pub(crate) fn clear(&mut self) { + // Client keeps the original unnamed Parse for later requests; this + // request's copy may have been rewritten for a particular backend. // We drop `last_parse` once the client has executed it. The gate is // `is_executable` (Bind/Execute/Query present), not the presence of // Sync: lib/pq sends Parse, Describe, Sync to learn parameter/row From 3d82bda9199157cc1dce97ea4b602f6dd04e5e16 Mon Sep 17 00:00:00 2001 From: Dipesh Babu Date: Mon, 21 Sep 2026 04:52:43 -0600 Subject: [PATCH 4/5] refactor(protocol): streamline SQL EXECUTE rewrites --- .../rust/tests/integration/simple_prepared.rs | 1 - pgdog/src/frontend/client_request.rs | 2 - .../router/parser/rewrite/statement/offset.rs | 5 ++- .../router/parser/rewrite/statement/plan.rs | 45 ++++++++++--------- 4 files changed, 28 insertions(+), 25 deletions(-) diff --git a/integration/rust/tests/integration/simple_prepared.rs b/integration/rust/tests/integration/simple_prepared.rs index e76d9cd62..384d54eab 100644 --- a/integration/rust/tests/integration/simple_prepared.rs +++ b/integration/rust/tests/integration/simple_prepared.rs @@ -34,7 +34,6 @@ async fn test_simple_prepared_ttl() { } /// -/// TODO: will need to support extended-protocol `Bind`s later for the re-write #[tokio::test] async fn test_simple_prepared_limit() { let mut conn = diff --git a/pgdog/src/frontend/client_request.rs b/pgdog/src/frontend/client_request.rs index 30b96bbf9..bec477928 100644 --- a/pgdog/src/frontend/client_request.rs +++ b/pgdog/src/frontend/client_request.rs @@ -82,8 +82,6 @@ impl ClientRequest { /// Remove any saved state from the request. pub(crate) fn clear(&mut self) { - // Client keeps the original unnamed Parse for later requests; this - // request's copy may have been rewritten for a particular backend. // We drop `last_parse` once the client has executed it. The gate is // `is_executable` (Bind/Execute/Query present), not the presence of // Sync: lib/pq sends Parse, Describe, Sync to learn parameter/row diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs b/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs index 380ac24c2..6d3dc68af 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/offset.rs @@ -168,8 +168,9 @@ impl OffsetPlan { let new_execute = new_execute.deref(); let new_execute_sql = deparse(new_execute)?; - // SQL EXECUTE can also arrive through the extended protocol. Update its - // internal Parse so cached outer statements receive the current limits. + // Replace SQL EXECUTE's LIMIT/OFFSET arguments with the values each shard + // needs (limit + offset, 0). Update Query for simple protocol and Parse + // or EnsureParsed for extended protocol, including cached statements. for message in request.messages.iter_mut() { match message { ProtocolMessage::Query(query) => query.set_query(new_execute_sql.as_str()), diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs index 1bbddf435..1e529118a 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs @@ -252,10 +252,15 @@ impl RewritePlan { anonymous_client_params = self.apply_parse(parse); } ProtocolMessage::Query(query) => self.apply_query(query).await?, - ProtocolMessage::Bind(bind) if self.prepare_rewrites.is_empty() => { - // SQL PREPARE's placeholders belong to the inner statement. - // SQL EXECUTE already has its generated values in the SQL text. - self.apply_bind(bind, timezone, timestamps).await? + ProtocolMessage::Bind(bind) => { + // Only ordinary statements need generated values appended to Bind. + // A nonempty prepare_rewrites means SQL PREPARE/EXECUTE: in + // `PREPARE foo AS INSERT INTO t VALUES ($1)`, $1 is supplied by + // a later EXECUTE, not this Bind. EXECUTE's generated values + // are already written into its SQL arguments. + if self.prepare_rewrites.is_empty() { + self.apply_bind(bind, timezone, timestamps).await? + } } _ => {} } @@ -274,28 +279,28 @@ impl RewritePlan { .prepare_rewrites .iter() .any(|rewrite| matches!(rewrite, PrepareExecute::Execute(_))) - && let Some(bind_index) = request - .messages - .iter() - .position(|message| matches!(message, ProtocolMessage::Bind(_))) && let Some(BufferedQuery::Prepared(mut parse)) = request.query()? { - // A cached outer EXECUTE can contain per-execution values and shard- - // dependent LIMIT/OFFSET. Parse it again without adding cache entries. - // Keep the original Bind name for the cross-shard result decoder. - parse.anonymize(); - request.anonymous_client_params = self.apply_parse(&mut parse); - for message in &mut request.messages { + let mut bind_index = None; + for (index, message) in request.messages.iter_mut().enumerate() { if let ProtocolMessage::Bind(bind) = message { + bind_index.get_or_insert(index); + // Keep the original Bind name for the cross-shard result decoder. *message = ProtocolMessage::BindAnonymous(bind.clone()); } } - request.last_parse = None; - // EnsurePrepared is a simple Query and destroys unnamed statements, - // so the internal Parse must follow it and precede Bind. - request - .messages - .insert(bind_index, ProtocolMessage::EnsureParsed(parse)); + if let Some(bind_index) = bind_index { + // A cached outer EXECUTE can contain per-execution values and shard- + // dependent LIMIT/OFFSET. Parse it again without adding cache entries. + parse.anonymize(); + request.anonymous_client_params = self.apply_parse(&mut parse); + request.last_parse = None; + // EnsurePrepared is a simple Query and destroys unnamed statements, + // so the internal Parse must follow it and precede Bind. + request + .messages + .insert(bind_index, ProtocolMessage::EnsureParsed(parse)); + } } self.apply_after_messages(request) From 06baddd5f1d360e00ee9f2d5501acd740c2d7db2 Mon Sep 17 00:00:00 2001 From: Dipesh Babu Date: Mon, 21 Sep 2026 04:56:17 -0600 Subject: [PATCH 5/5] style(protocol): explain SQL Bind handling above match guard --- .../router/parser/rewrite/statement/plan.rs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs index 1e529118a..a0ac6ed4a 100644 --- a/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs +++ b/pgdog/src/frontend/router/parser/rewrite/statement/plan.rs @@ -252,15 +252,13 @@ impl RewritePlan { anonymous_client_params = self.apply_parse(parse); } ProtocolMessage::Query(query) => self.apply_query(query).await?, - ProtocolMessage::Bind(bind) => { - // Only ordinary statements need generated values appended to Bind. - // A nonempty prepare_rewrites means SQL PREPARE/EXECUTE: in - // `PREPARE foo AS INSERT INTO t VALUES ($1)`, $1 is supplied by - // a later EXECUTE, not this Bind. EXECUTE's generated values - // are already written into its SQL arguments. - if self.prepare_rewrites.is_empty() { - self.apply_bind(bind, timezone, timestamps).await? - } + // Only ordinary statements need generated values appended to Bind. + // A nonempty prepare_rewrites means SQL PREPARE/EXECUTE: in + // `PREPARE foo AS INSERT INTO t VALUES ($1)`, $1 is supplied by + // a later EXECUTE, not this Bind. EXECUTE's generated values + // are already written into its SQL arguments. + ProtocolMessage::Bind(bind) if self.prepare_rewrites.is_empty() => { + self.apply_bind(bind, timezone, timestamps).await? } _ => {} }