diff --git a/integration/ci/apt.sh b/integration/ci/apt.sh index f3f289ea5..9a77f25e1 100755 --- a/integration/ci/apt.sh +++ b/integration/ci/apt.sh @@ -1,12 +1,11 @@ #!/usr/bin/env bash -# Thin `apt-get install` wrapper that no-ops on non-Linux platforms (macOS -# developers running these scripts locally already have the deps installed -# via their own package manager). +# Thin `apt-get install` wrapper used in CI only. +# No-ops when CI is not set (local dev) or apt-get is unavailable. # # Usage: integration/ci/apt.sh ... set -euo pipefail -if [[ "$(uname -s)" != "Linux" ]]; then +if [[ -z "${CI:-}" ]] || ! command -v apt-get &>/dev/null; then exit 0 fi diff --git a/integration/prepared_statements_full/users.toml b/integration/prepared_statements_full/users.toml deleted file mode 100644 index 9a8205f04..000000000 --- a/integration/prepared_statements_full/users.toml +++ /dev/null @@ -1,4 +0,0 @@ -[[users]] -database = "pgdog" -name = "pgdog" -password = "pgdog" diff --git a/integration/ruby/common.sh b/integration/ruby/common.sh new file mode 100644 index 000000000..5abe24a94 --- /dev/null +++ b/integration/ruby/common.sh @@ -0,0 +1,47 @@ +#!/bin/bash +# +# Shared helpers for ruby integration suites. +# Source this file; do not execute directly. +# +RUBY_COMMON_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source "${RUBY_COMMON_DIR}/../common.sh" + +# Install system packages and the bundler gem. Call once per CI run. +function install_deps() { + # Native gem extensions (psych, pg) need yaml + libpq headers. + bash ${RUBY_COMMON_DIR}/../ci/apt.sh ruby-dev libyaml-dev libpq-dev build-essential + command -v bundle >/dev/null || sudo gem install bundler --no-document +} + +# Run bundle install and rspec in TARGET_DIR using the shared Gemfile. +# Defaults to RUBY_COMMON_DIR when called with no argument. +function dev_suite() { + local target_dir="${1:-$RUBY_COMMON_DIR}" + + export BUNDLE_GEMFILE="${RUBY_COMMON_DIR}/Gemfile" + export GEM_HOME=~/.gem + mkdir -p ${GEM_HOME} + + pushd "${target_dir}" + bundle install + bundle exec rspec *_spec.rb + popd +} + +# Full CI cycle for a single suite: start pgdog, run tests, stop. +# CONFIG_DIR is optional — omit to use the default integration/ config. +# Call install_deps before the first run_suite in a process. +function run_suite() { + local config_dir="${1:-}" + + if [ -n "$config_dir" ]; then + run_pgdog "$config_dir" + else + run_pgdog + fi + wait_for_pgdog + + dev_suite "$config_dir" + + stop_pgdog +} diff --git a/integration/ruby/dev.sh b/integration/ruby/dev.sh index f36274590..14d01caff 100644 --- a/integration/ruby/dev.sh +++ b/integration/ruby/dev.sh @@ -1,12 +1,5 @@ #!/bin/bash set -e SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) - -pushd ${SCRIPT_DIR} - -export GEM_HOME=~/.gem -mkdir -p ${GEM_HOME} -bundle install -bundle exec rspec *_spec.rb - -popd +source "${SCRIPT_DIR}/common.sh" +dev_suite diff --git a/integration/ruby/prepared_disabled/dev.sh b/integration/ruby/prepared_disabled/dev.sh new file mode 100755 index 000000000..31c74fa4c --- /dev/null +++ b/integration/ruby/prepared_disabled/dev.sh @@ -0,0 +1,5 @@ +#!/bin/bash +set -e +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source "${SCRIPT_DIR}/../common.sh" +dev_suite "$SCRIPT_DIR" diff --git a/integration/ruby/prepared_disabled/pgdog.toml b/integration/ruby/prepared_disabled/pgdog.toml new file mode 100644 index 000000000..e133d29ec --- /dev/null +++ b/integration/ruby/prepared_disabled/pgdog.toml @@ -0,0 +1,10 @@ +[general] +prepared_statements = "disabled" +default_pool_size = 2 + +[[databases]] +name = "pgdog" +host = "127.0.0.1" + +[admin] +password = "pgdog" diff --git a/integration/ruby/prepared_disabled/prepared_spec.rb b/integration/ruby/prepared_disabled/prepared_spec.rb new file mode 100644 index 000000000..2b73f8083 --- /dev/null +++ b/integration/ruby/prepared_disabled/prepared_spec.rb @@ -0,0 +1,112 @@ +# frozen_string_literal: true + +require_relative '../rspec_helper' + +# With prepared_statements = "disabled" pgdog forwards protocol messages as-is +# without rewriting or caching. +describe 'prepared_statements = disabled' do + after { ensure_done } + + # 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 + conn = connect + 10.times do |i| + res = conn.exec_params('SELECT $1::bigint * 2 AS val', [i]) + expect(res[0]['val'].to_i).to eq(i * 2) + end + conn.close + end + + # Session mode pins one backend for the connection lifetime; pass-through + # is sufficient because prepare and execute always reach the same backend. + it 'passes named statements through in session mode' do + conn = connect('pgdog', 'pgdog_session') + conn.prepare('session_stmt', 'SELECT $1::bigint AS val') + 10.times do |i| + res = conn.exec_prepared('session_stmt', [i]) + expect(res[0]['val'].to_i).to eq(i) + end + conn.close + end + + # Session mode gives each client its own dedicated backend, so two session + # connections are guaranteed to land on different backends. Without a global + # cache the execute on conn2 reaches a backend that never saw the prepare. + it 'does not share statements across connections' do + conn1 = connect('pgdog', 'pgdog_session') + conn2 = connect('pgdog', 'pgdog_session') + conn1.prepare('cross_stmt', 'SELECT $1::bigint AS val') + expect do + conn2.exec_prepared('cross_stmt', [7]) + end.to raise_error(PG::Error) + conn1.close + conn2.close + end + + # In transaction pool mode each query can land on a different backend. + # disabled mode forwards Parse and Bind as-is with no global cache, so a + # Bind that arrives on a backend that never saw the Parse fails. + # + # Sequential tests cannot force a crossing: a single connection always + # returns the backend to the LIFO top between queries, so the next query + # gets the same backend. Concurrent threads make the pool hand out both + # backends simultaneously. With 5 threads and pool_size = 2, the pigeonhole + # principle guarantees that at least 3 threads will attempt PREPARE on a + # backend that already holds the statement, producing 'already exists' + # errors — regardless of timing. + # + # Mirror: full/'executes named extended-protocol statements in + # transaction pool mode' — same structure, opposite expectation. + it 'fails named extended-protocol statements in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 5.times.map do + Thread.new do + conn = connect + begin + conn.prepare('ext_stmt', 'SELECT $1::bigint AS val') + 20.times { conn.exec_prepared('ext_stmt', [42]) } + rescue PG::Error => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).not_to be_empty + end + + # Same mechanism as the extended-protocol test above, but for the + # simple-protocol PREPARE/EXECUTE path. disabled mode forwards the SQL + # statement text as-is, so EXECUTE on a backend that never saw the + # PREPARE fails with 'prepared statement does not exist' or, if two + # threads hit the same backend, 'already exists'. + # + # Mirror: full/'rewrites simple-protocol PREPARE / EXECUTE in + # transaction pool mode' — same structure, opposite expectation. + it 'fails SQL PREPARE/EXECUTE in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 5.times.map do + Thread.new do + conn = connect + begin + conn.exec('PREPARE sql_stmt AS SELECT $1::bigint * 2') + 20.times { |i| conn.exec("EXECUTE sql_stmt(#{i})") } + rescue PG::Error => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).not_to be_empty + end +end diff --git a/integration/ruby/prepared_disabled/run.sh b/integration/ruby/prepared_disabled/run.sh new file mode 100755 index 000000000..dc3e993eb --- /dev/null +++ b/integration/ruby/prepared_disabled/run.sh @@ -0,0 +1,6 @@ +#!/bin/bash +set -e +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source "${SCRIPT_DIR}/../common.sh" +install_deps +run_suite "$SCRIPT_DIR" diff --git a/integration/ruby/prepared_disabled/users.toml b/integration/ruby/prepared_disabled/users.toml new file mode 100644 index 000000000..6ecd9dce0 --- /dev/null +++ b/integration/ruby/prepared_disabled/users.toml @@ -0,0 +1,12 @@ +[[users]] +database = "pgdog" +name = "pgdog" +password = "pgdog" + + +[[users]] +name = "pgdog_session" +database = "pgdog" +password = "pgdog" +server_user = "pgdog" +pooler_mode = "session" diff --git a/integration/ruby/prepared_full/dev.sh b/integration/ruby/prepared_full/dev.sh new file mode 100755 index 000000000..31c74fa4c --- /dev/null +++ b/integration/ruby/prepared_full/dev.sh @@ -0,0 +1,5 @@ +#!/bin/bash +set -e +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source "${SCRIPT_DIR}/../common.sh" +dev_suite "$SCRIPT_DIR" diff --git a/integration/prepared_statements_full/pgdog.toml b/integration/ruby/prepared_full/pgdog.toml similarity index 84% rename from integration/prepared_statements_full/pgdog.toml rename to integration/ruby/prepared_full/pgdog.toml index 37ae8403d..77ec881eb 100644 --- a/integration/prepared_statements_full/pgdog.toml +++ b/integration/ruby/prepared_full/pgdog.toml @@ -1,5 +1,6 @@ [general] prepared_statements = "full" +default_pool_size = 2 [[databases]] name = "pgdog" diff --git a/integration/ruby/prepared_full/prepared_spec.rb b/integration/ruby/prepared_full/prepared_spec.rb new file mode 100644 index 000000000..69defacb5 --- /dev/null +++ b/integration/ruby/prepared_full/prepared_spec.rb @@ -0,0 +1,105 @@ +# frozen_string_literal: true + +require_relative '../rspec_helper' + +describe 'prepared_statements = full' do + after { ensure_done } + + # 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 + conn = connect + 10.times do |i| + res = conn.exec_params('SELECT $1::bigint * 2 AS val', [i]) + expect(res[0]['val'].to_i).to eq(i * 2) + end + conn.close + end + + # Mirror of disabled suite: session mode pins the client to one backend for + # the connection lifetime, so named statements always reach the backend that + # holds them regardless of the prepared_statements setting. + it 'passes named statements in session mode' do + conn = connect('pgdog', 'pgdog_session') + conn.prepare('session_stmt', 'SELECT $1::bigint AS val') + 10.times do |i| + res = conn.exec_prepared('session_stmt', [i]) + expect(res[0]['val'].to_i).to eq(i) + end + conn.close + end + + # Mirror of disabled/'fails SQL PREPARE/EXECUTE in transaction pool mode'. + # full mode intercepts PREPARE, renames the statement to an internal name + # (__pgdog_N), and replays the PREPARE on any backend that hasn't seen it + # before executing. 5 threads × 20 iterations with pool_size = 2 generates + # the same backend crossings as the disabled test, but all succeed. + it 'rewrites simple-protocol PREPARE / EXECUTE in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 5.times.map do + Thread.new do + conn = connect + begin + conn.exec('PREPARE sql_stmt AS SELECT $1::bigint * 2') + 20.times { |i| conn.exec("EXECUTE sql_stmt(#{i})") } + rescue PG::Error => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).to be_empty + end + + # Session mode gives each client its own dedicated backend, so two session + # connections are guaranteed to land on different backends. Without a global + # cache the execute on conn2 reaches a backend that never saw the prepare. + it 'does not share statements across connections' do + conn1 = connect('pgdog', 'pgdog_session') + conn2 = connect('pgdog', 'pgdog_session') + conn1.prepare('cross_stmt', 'SELECT $1::bigint AS val') + expect do + conn2.exec_prepared('cross_stmt', [7]) + end.to raise_error(PG::Error) + conn1.close + conn2.close + end + + # Mirror of disabled/'fails named extended-protocol statements in + # transaction pool mode'. full mode renames each frontend's Parse to an + # internal name (__pgdog_N, unique per frontend) and replays it on any + # backend before the Bind. 5 threads × 20 iterations with pool_size = 2 + # forces genuine crossings \ the replay ensures all succeed. + # Result values are also verified to guard against silent data corruption. + it 'executes named extended-protocol statements in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 5.times.map do + Thread.new do + conn = connect + begin + conn.prepare('ext_stmt', 'SELECT $1::bigint AS val') + 20.times do |i| + res = conn.exec_prepared('ext_stmt', [i]) + val = res[0]['val'].to_i + raise "expected #{i}, got #{val}" unless val == i + end + rescue => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).to be_empty + end + +end diff --git a/integration/ruby/prepared_full/run.sh b/integration/ruby/prepared_full/run.sh new file mode 100755 index 000000000..dc3e993eb --- /dev/null +++ b/integration/ruby/prepared_full/run.sh @@ -0,0 +1,6 @@ +#!/bin/bash +set -e +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source "${SCRIPT_DIR}/../common.sh" +install_deps +run_suite "$SCRIPT_DIR" diff --git a/integration/ruby/prepared_full/users.toml b/integration/ruby/prepared_full/users.toml new file mode 100644 index 000000000..6ecd9dce0 --- /dev/null +++ b/integration/ruby/prepared_full/users.toml @@ -0,0 +1,12 @@ +[[users]] +database = "pgdog" +name = "pgdog" +password = "pgdog" + + +[[users]] +name = "pgdog_session" +database = "pgdog" +password = "pgdog" +server_user = "pgdog" +pooler_mode = "session" diff --git a/integration/ruby/prepared_spec.rb b/integration/ruby/prepared_spec.rb new file mode 100644 index 000000000..be23e1aa2 --- /dev/null +++ b/integration/ruby/prepared_spec.rb @@ -0,0 +1,107 @@ +# frozen_string_literal: true + +require_relative 'rspec_helper' + +# Uses the main integration pgdog.toml which sets prepared_statements = "extended". +# "extended" rewrites and replays named extended-protocol statements (Parse/Bind) +# across backends, but does NOT intercept SQL PREPARE / EXECUTE. + + +describe 'prepared_statements = extended' do + after { ensure_done } + + # 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 + conn = connect + 10.times do |i| + res = conn.exec_params('SELECT $1::bigint * 2 AS val', [i]) + expect(res[0]['val'].to_i).to eq(i * 2) + end + conn.close + end + + # Session mode pins one backend for the connection lifetime; pass-through + # is sufficient because prepare and execute always reach the same backend. + it 'passes named statements in session mode' do + conn = connect('pgdog', 'pgdog_session') + conn.prepare('session_stmt', 'SELECT $1::bigint AS val') + 10.times do |i| + res = conn.exec_prepared('session_stmt', [i]) + expect(res[0]['val'].to_i).to eq(i) + end + conn.close + end + + # The rewrite is per-frontend (per client connection). conn2 has no mapping + # for 'cross_stmt', so its Bind is forwarded with the original name to a + # backend that never saw the Parse for that name. + it 'does not share statements across connections' do + conn1 = connect('pgdog', 'pgdog_session') + conn2 = connect('pgdog', 'pgdog_session') + conn1.prepare('cross_stmt', 'SELECT $1::bigint AS val') + expect do + conn2.exec_prepared('cross_stmt', [7]) + end.to raise_error(PG::Error) + conn1.close + conn2.close + end + + # extended mode renames each frontend's Parse to an internal name + # (__pgdog_N, unique per frontend) and replays it on any backend before + # the Bind. 15 threads × 20 iterations with a default pool of 10 guarantees + # genuine crossings via the pigeonhole principle; the replay ensures all succeed. + # Result values are verified to guard against silent data corruption. + it 'executes named extended-protocol statements in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 15.times.map do + Thread.new do + conn = connect + begin + conn.prepare('ext_stmt', 'SELECT $1::bigint AS val') + 20.times do |i| + res = conn.exec_prepared('ext_stmt', [i]) + val = res[0]['val'].to_i + raise "expected #{i}, got #{val}" unless val == i + end + rescue => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).to be_empty + end + + # extended mode does NOT intercept SQL PREPARE / EXECUTE — those are + # forwarded as-is. With 15 threads and a default pool of 10, the pigeonhole + # principle guarantees crossings: at least 5 threads hit a backend that + # already holds 'sql_stmt' ('already exists') or one that never saw the + # PREPARE ('does not exist'). Either way, errors accumulate. + it 'fails SQL PREPARE/EXECUTE in transaction pool mode' do + errors = [] + mutex = Mutex.new + + threads = 15.times.map do + Thread.new do + conn = connect + begin + conn.exec('PREPARE sql_stmt AS SELECT $1::bigint * 2') + 20.times { |i| conn.exec("EXECUTE sql_stmt(#{i})") } + rescue PG::Error => e + mutex.synchronize { errors << e.message } + ensure + conn.close rescue nil + end + end + end + + threads.each(&:join) + expect(errors).not_to be_empty + end +end diff --git a/integration/ruby/rspec_helper.rb b/integration/ruby/rspec_helper.rb index 8a7cf2990..f0c0015d0 100644 --- a/integration/ruby/rspec_helper.rb +++ b/integration/ruby/rspec_helper.rb @@ -50,22 +50,67 @@ def admin_stats(database, column = nil) stats end +def connect(db = 'pgdog', user = 'pgdog') + PG.connect(dbname: db, user: user, password: 'pgdog', port: 6432, host: '127.0.0.1') +end + def ensure_done - conn = PG.connect(dbname: 'admin', user: 'admin', password: 'pgdog', port: 6432, host: '127.0.0.1') - pools = conn.exec 'SHOW POOLS' + deadline = Time.now + 2 + pools = [] + clients = [] + servers = [] + pg_clients = [] + current_client_id = nil + + loop do + conn = PG.connect(dbname: 'admin', user: 'admin', password: 'pgdog', port: 6432, host: '127.0.0.1') + begin + pools = conn.exec 'SHOW POOLS' + current_client_id = conn.backend_pid + clients = conn.exec 'SHOW CLIENTS' + servers = conn.exec 'SHOW SERVERS' + ensure + conn.close + end + + pg_conn = PG.connect(dbname: 'pgdog', user: 'pgdog', password: 'pgdog', port: 5432, host: '127.0.0.1') + begin + pg_clients = pg_conn.exec 'SELECT state FROM pg_stat_activity'\ + " WHERE datname IN ('pgdog', 'shard_0', 'shard_1')"\ + " AND backend_type = 'client backend'"\ + " AND query NOT LIKE '%pg_stat_activity%'" + ensure + pg_conn.close + end + + pools_ready = pools.all? do |pool| + pool['sv_active'] == '0' && pool['cl_waiting'] == '0' && pool['out_of_sync'] == '0' + end + clients_ready = clients.all? do |client| + client['id'].to_i == current_client_id || client['state'] == 'idle' + end + servers_ready = servers + .select { |server| server['application_name'] != 'PgDog Pub/Sub Listener' } + .all? { |server| server['state'] == 'idle' } + pg_clients_ready = pg_clients.all? { |client| client['state'] == 'idle' } + + break if pools_ready && clients_ready && servers_ready && pg_clients_ready + break if Time.now >= deadline + + sleep 0.05 + end + pools.each do |pool| expect(pool['sv_active']).to eq('0') expect(pool['cl_waiting']).to eq('0') expect(pool['out_of_sync']).to eq('0') end - current_client_id = conn.backend_pid - clients = conn.exec 'SHOW CLIENTS' + clients.each do |client| next if client['id'].to_i == current_client_id - expect(client['state']).to eq('idle') end - servers = conn.exec 'SHOW SERVERS' + servers .select do |server| server['application_name'] != 'PgDog Pub/Sub Listener' @@ -74,12 +119,7 @@ def ensure_done expect(server['state']).to eq('idle') end - conn = PG.connect(dbname: 'pgdog', user: 'pgdog', password: 'pgdog', port: 5432, host: '127.0.0.1') - clients = conn.exec 'SELECT state FROM pg_stat_activity'\ - " WHERE datname IN ('pgdog', 'shard_0', 'shard_1')"\ - " AND backend_type = 'client backend'"\ - " AND query NOT LIKE '%pg_stat_activity%'" - clients.each do |client| + pg_clients.each do |client| expect(client['state']).to eq('idle') end end diff --git a/integration/ruby/run.sh b/integration/ruby/run.sh index 792b35971..e827d6857 100644 --- a/integration/ruby/run.sh +++ b/integration/ruby/run.sh @@ -1,15 +1,10 @@ #!/bin/bash set -e SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) -source ${SCRIPT_DIR}/../common.sh +source "${SCRIPT_DIR}/common.sh" -# Native gem extensions (psych, pg) need yaml + libpq headers. -bash ${SCRIPT_DIR}/../ci/apt.sh ruby-dev libyaml-dev libpq-dev build-essential -command -v bundle >/dev/null || sudo gem install bundler --no-document +install_deps -run_pgdog -wait_for_pgdog - -bash ${SCRIPT_DIR}/dev.sh - -stop_pgdog +run_suite +run_suite "${SCRIPT_DIR}/prepared_disabled" +run_suite "${SCRIPT_DIR}/prepared_full" diff --git a/pgdog-config/src/pooling.rs b/pgdog-config/src/pooling.rs index 3fbef5207..6d15b24b3 100644 --- a/pgdog-config/src/pooling.rs +++ b/pgdog-config/src/pooling.rs @@ -31,6 +31,15 @@ impl PreparedStatements { pub fn rewrite_anonymous(&self) -> bool { matches!(self, PreparedStatements::ExtendedAnonymous) } + + pub fn handles_extended(&self) -> bool { + matches!( + self, + PreparedStatements::Extended + | PreparedStatements::ExtendedAnonymous + | PreparedStatements::Full + ) + } } impl FromStr for PreparedStatements { diff --git a/pgdog/src/frontend/client/query_engine/rewrite.rs b/pgdog/src/frontend/client/query_engine/rewrite.rs index 907b81922..a8e143694 100644 --- a/pgdog/src/frontend/client/query_engine/rewrite.rs +++ b/pgdog/src/frontend/client/query_engine/rewrite.rs @@ -1,7 +1,4 @@ -use crate::{ - config::PreparedStatements, - frontend::router::parser::{AstContext, Cache}, -}; +use crate::frontend::router::parser::{AstContext, Cache}; use super::*; @@ -14,12 +11,8 @@ impl QueryEngine { for message in context.client_request.iter_mut() { if message.is_extended() { let level = context.prepared_statements.level; - match (level, message.anonymous()) { - (PreparedStatements::ExtendedAnonymous, _) - | (PreparedStatements::Extended, false) => { - context.prepared_statements.maybe_rewrite(message)? - } - _ => (), + if level.handles_extended() && (level.rewrite_anonymous() || !message.anonymous()) { + context.prepared_statements.maybe_rewrite(message)?; } } }