fix(sql): send Parse+Bind+Execute atomically for unnamed prepared statements

src/sql/postgres/PostgresRequest.zig

@@ -257,6 +257,65 @@ pub fn bindAndExecute(
     try writer.write(&protocol.Sync);
 }
 
+/// Atomically sends Parse + [Describe] + Bind + Execute + Flush + Sync as a single message batch.
+/// This is required for unnamed prepared statements to work correctly with connection poolers
+/// like PgBouncer in transaction mode, which may reassign server connections between protocol
+/// round-trips. Without this, Parse and Bind+Execute could be routed to different backend
+/// connections, causing queries to execute against the wrong prepared statement.
+pub fn parseAndBindAndExecute(
+    globalObject: *jsc.JSGlobalObject,
+    query: []const u8,
+    statement: *PostgresSQLStatement,
+    array_value: JSValue,
+    columns_value: JSValue,
+    include_describe: bool,
+    comptime Context: type,
+    writer: protocol.NewWriter(Context),
+) AnyPostgresError!void {
+    const name = statement.signature.prepared_statement_name;
+
+    // Parse
+    {
+        var q = protocol.Parse{
+            .name = name,
+            .params = statement.signature.fields,
+            .query = query,
+        };
+        try q.writeInternal(Context, writer);
+        debug("Parse: {f}", .{bun.fmt.quote(query)});
+    }
+
+    // Describe (needed on first execution to learn parameter/result types for caching)
+    if (include_describe) {
+        var d = protocol.Describe{
+            .p = .{
+                .prepared_statement = name,
+            },
+        };
+        try d.writeInternal(Context, writer);
+        debug("Describe: {f}", .{bun.fmt.quote(name)});
+    }
+
+    // Bind — use server-provided types if available (binary format), otherwise
+    // fall back to signature types (text format for unknowns). The server will
+    // handle text-to-type conversion based on the parameter types from Parse.
+    const param_fields = if (statement.parameters.len > 0) statement.parameters else statement.signature.fields;
+    const result_fields = statement.fields;
+
+    try writeBind(name, bun.String.empty, globalObject, array_value, columns_value, param_fields, result_fields, Context, writer);
+
+    // Execute
+    var exec = protocol.Execute{
+        .p = .{
+            .prepared_statement = name,
+        },
+    };
+    try exec.writeInternal(Context, writer);
+
+    try writer.write(&protocol.Flush);
+    try writer.write(&protocol.Sync);
+}
+
 pub fn executeQuery(
     query: []const u8,
     comptime Context: type,

src/sql/postgres/PostgresSQLConnection.zig

@@ -1090,6 +1090,16 @@ pub fn canPrepareQuery(noalias this: *const @This()) bool {
     return this.flags.is_ready_for_query and !this.flags.waiting_to_prepare and this.pipelined_requests == 0;
 }
 
+/// Process pending requests and flush. Called from the enqueue path when
+/// unnamed prepared statements with params skip writeQuery+Sync and need
+/// advance() to send everything atomically on an idle connection.
+pub fn advanceAndFlush(this: *PostgresSQLConnection) void {
+    if (!this.flags.has_backpressure and this.flags.is_ready_for_query) {
+        this.advance();
+        this.flushData();
+    }
+}
+
 fn advance(this: *PostgresSQLConnection) void {
     var offset: usize = 0;
     debug("advance", .{});
@@ -1182,24 +1192,52 @@ fn advance(this: *PostgresSQLConnection) void {
                                 const binding_value = PostgresSQLQuery.js.bindingGetCached(thisValue) orelse .zero;
                                 const columns_value = PostgresSQLQuery.js.columnsGetCached(thisValue) orelse .zero;
                                 req.flags.binary = statement.fields.len > 0;
-                                debug("binding and executing stmt", .{});
-                                PostgresRequest.bindAndExecute(this.globalObject, statement, binding_value, columns_value, PostgresSQLConnection.Writer, this.writer()) catch |err| {
-                                    if (this.globalObject.tryTakeException()) |err_| {
-                                        req.onJSError(err_, this.globalObject);
-                                    } else {
-                                        req.onWriteFail(err, this.globalObject, this.getQueriesArray());
-                                    }
-                                    if (offset == 0) {
-                                        req.deref();
-                                        this.requests.discard(1);
-                                    } else {
-                                        // deinit later
-                                        req.status = .fail;
-                                        offset += 1;
-                                    }
-                                    debug("bind and execute failed: {s}", .{@errorName(err)});
-                                    continue;
-                                };
+
+                                if (this.flags.use_unnamed_prepared_statements) {
+                                    // For unnamed prepared statements, always include Parse
+                                    // before Bind+Execute. The unnamed statement may not exist
+                                    // on the current server connection when using PgBouncer or
+                                    // other connection poolers in transaction mode.
+                                    debug("parse, bind and execute unnamed stmt", .{});
+                                    var query_str = req.query.toUTF8(bun.default_allocator);
+                                    defer query_str.deinit();
+                                    PostgresRequest.parseAndBindAndExecute(this.globalObject, query_str.slice(), statement, binding_value, columns_value, false, PostgresSQLConnection.Writer, this.writer()) catch |err| {
+                                        if (this.globalObject.tryTakeException()) |err_| {
+                                            req.onJSError(err_, this.globalObject);
+                                        } else {
+                                            req.onWriteFail(err, this.globalObject, this.getQueriesArray());
+                                        }
+                                        if (offset == 0) {
+                                            req.deref();
+                                            this.requests.discard(1);
+                                        } else {
+                                            // deinit later
+                                            req.status = .fail;
+                                            offset += 1;
+                                        }
+                                        debug("parse, bind and execute failed: {s}", .{@errorName(err)});
+                                        continue;
+                                    };
+                                } else {
+                                    debug("binding and executing stmt", .{});
+                                    PostgresRequest.bindAndExecute(this.globalObject, statement, binding_value, columns_value, PostgresSQLConnection.Writer, this.writer()) catch |err| {
+                                        if (this.globalObject.tryTakeException()) |err_| {
+                                            req.onJSError(err_, this.globalObject);
+                                        } else {
+                                            req.onWriteFail(err, this.globalObject, this.getQueriesArray());
+                                        }
+                                        if (offset == 0) {
+                                            req.deref();
+                                            this.requests.discard(1);
+                                        } else {
+                                            // deinit later
+                                            req.status = .fail;
+                                            offset += 1;
+                                        }
+                                        debug("bind and execute failed: {s}", .{@errorName(err)});
+                                        continue;
+                                    };
+                                }
 
                                 this.flags.is_ready_for_query = false;
                                 req.status = .binding;
@@ -1268,6 +1306,49 @@ fn advance(this: *PostgresSQLConnection) void {
                                     return;
                                 }
 
+                                if (this.flags.use_unnamed_prepared_statements) {
+                                    // For unnamed prepared statements, send Parse+Describe+Bind+Execute
+                                    // atomically to prevent PgBouncer from splitting them across
+                                    // server connections. Uses signature field types for encoding
+                                    // (text format for unknowns); actual types will be cached from
+                                    // ParameterDescription for subsequent executions.
+                                    const thisValue = req.thisValue.tryGet() orelse {
+                                        bun.assertf(false, "query value was freed earlier than expected", .{});
+                                        bun.assert(offset == 0);
+                                        req.deref();
+                                        this.requests.discard(1);
+                                        continue;
+                                    };
+                                    const binding_value = PostgresSQLQuery.js.bindingGetCached(thisValue) orelse .zero;
+                                    const columns_value = PostgresSQLQuery.js.columnsGetCached(thisValue) orelse .zero;
+                                    debug("parseAndBindAndExecute (unnamed, first execution)", .{});
+                                    PostgresRequest.parseAndBindAndExecute(this.globalObject, query_str.slice(), statement, binding_value, columns_value, true, PostgresSQLConnection.Writer, this.writer()) catch |err| {
+                                        if (this.globalObject.tryTakeException()) |err_| {
+                                            req.onJSError(err_, this.globalObject);
+                                        } else {
+                                            statement.status = .failed;
+                                            statement.error_response = .{ .postgres_error = err };
+                                            req.onWriteFail(err, this.globalObject, this.getQueriesArray());
+                                        }
+                                        bun.assert(offset == 0);
+                                        req.deref();
+                                        this.requests.discard(1);
+                                        debug("parseAndBindAndExecute failed: {s}", .{@errorName(err)});
+                                        continue;
+                                    };
+                                    this.flags.is_ready_for_query = false;
+                                    this.flags.waiting_to_prepare = true;
+                                    req.status = .binding;
+                                    statement.status = .parsing;
+                                    req.flags.pipelined = true;
+                                    this.pipelined_requests += 1;
+                                    this.flushDataAndResetTimeout();
+                                    return;
+                                }
+
+                                // Named prepared statements: send Parse+Describe first, wait for
+                                // ParameterDescription, then send Bind+Execute in a second phase.
+                                // This is safe because named statements persist on the connection.
                                 const connection_writer = this.writer();
                                 debug("writing query", .{});
                                 // write query and wait for it to be prepared

src/sql/postgres/PostgresSQLQuery.zig

@@ -420,7 +420,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra
                 this.status = .binding;
                 did_write = true;
                 connection.flags.waiting_to_prepare = true;
-            } else {
+            } else if (!connection.flags.use_unnamed_prepared_statements) {
+                // Named prepared statements: send Parse+Describe+Sync now and wait
+                // for ParameterDescription before sending Bind+Execute in advance().
                 debug("writeQuery", .{});
 
                 PostgresRequest.writeQuery(query_str.slice(), signature.prepared_statement_name, signature.fields, PostgresSQLConnection.Writer, writer) catch |err| {
@@ -450,6 +452,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra
                 did_write = true;
                 connection.flags.waiting_to_prepare = true;
             }
+            // Unnamed prepared statements with params: skip writeQuery+Sync here.
+            // advance() will send Parse+Describe+Bind+Execute atomically via
+            // parseAndBindAndExecute(), preventing PgBouncer from splitting them.
         }
         {
             const stmt = bun.default_allocator.create(PostgresSQLStatement) catch {
@@ -465,15 +470,15 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra
                 stmt.* = .{
                     .signature = signature,
                     .ref_count = .initExactRefs(2),
-                    .status = if (can_execute) .parsing else .pending,
+                    .status = if (did_write) .parsing else .pending,
                 };
                 this.statement = stmt;
 
                 entry_value.* = stmt;
             } else {
                 stmt.* = .{
                     .signature = signature,
-                    .status = if (can_execute) .parsing else .pending,
+                    .status = if (did_write) .parsing else .pending,
                 };
                 this.statement = stmt;
             }
@@ -488,6 +493,9 @@ pub fn doRun(this: *PostgresSQLQuery, globalObject: *jsc.JSGlobalObject, callfra
         connection.flushDataAndResetTimeout();
     } else {
         connection.resetConnectionTimeout();
+        // For unnamed prepared statements with params, we skip writeQuery+Sync
+        // in the enqueue path and let advance() handle it atomically.
+        connection.advanceAndFlush();
     }
     return .js_undefined;
 }

test/js/sql/sql-prepare-false.test.ts

@@ -0,0 +1,135 @@
+import { SQL } from "bun";
+import { afterAll, describe, expect, test } from "bun:test";
+import * as dockerCompose from "../../docker/index.ts";
+
+// Tests for `prepare: false` (unnamed prepared statements).
+// These verify that parameterized queries work correctly when using unnamed
+// prepared statements, which is critical for PgBouncer compatibility.
+
+describe("PostgreSQL prepare: false", async () => {
+  let container: { port: number; host: string };
+
+  try {
+    const info = await dockerCompose.ensure("postgres_plain");
+    container = { port: info.ports[5432], host: info.host };
+  } catch (e) {
+    test.skip(`Docker not available: ${e}`);
+    return;
+  }
+
+  const options = {
+    db: "bun_sql_test",
+    username: "bun_sql_test",
+    host: container.host,
+    port: container.port,
+    max: 1,
+    prepare: false,
+  };
+
+  afterAll(async () => {
+    if (!process.env.BUN_KEEP_DOCKER) {
+      await dockerCompose.down();
+    }
+  });
+
+  test("basic parameterized query", async () => {
+    await using db = new SQL(options);
+    const [{ x }] = await db`SELECT ${42}::int AS x`;
+    expect(x).toBe(42);
+  });
+
+  test("multiple parameterized queries sequentially", async () => {
+    await using db = new SQL(options);
+
+    const [{ a }] = await db`SELECT ${1}::int AS a`;
+    expect(a).toBe(1);
+
+    const [{ b }] = await db`SELECT ${"hello"}::text AS b`;
+    expect(b).toBe("hello");
+
+    const [{ c }] = await db`SELECT ${3.14}::float8 AS c`;
+    expect(c).toBeCloseTo(3.14);
+  });
+
+  test("same query repeated with different params", async () => {
+    await using db = new SQL(options);
+    for (let i = 0; i < 10; i++) {
+      const [{ x }] = await db`SELECT ${i}::int AS x`;
+      expect(x).toBe(i);
+    }
+  });
+
+  test("concurrent queries with different tables return correct results", async () => {
+    // This test simulates the scenario where concurrent unnamed prepared
+    // statements could interfere with each other via PgBouncer.
+    await using db = new SQL({ ...options, max: 4 });
+
+    // Create real tables (not temp, so they're visible across connections)
+    await db`CREATE TABLE IF NOT EXISTS prepare_false_test_a (id int, val text)`;
+    await db`CREATE TABLE IF NOT EXISTS prepare_false_test_b (id int, val text)`;
+    await db`DELETE FROM prepare_false_test_a`;
+    await db`DELETE FROM prepare_false_test_b`;
+    await db`INSERT INTO prepare_false_test_a VALUES (1, 'from_a')`;
+    await db`INSERT INTO prepare_false_test_b VALUES (1, 'from_b')`;
+
+    // Run concurrent parameterized queries against different tables
+    const results = await Promise.all([
+      db`SELECT val FROM prepare_false_test_a WHERE id = ${1}`,
+      db`SELECT val FROM prepare_false_test_b WHERE id = ${1}`,
+      db`SELECT val FROM prepare_false_test_a WHERE id = ${1}`,
+      db`SELECT val FROM prepare_false_test_b WHERE id = ${1}`,
+    ]);
+
+    expect(results[0][0].val).toBe("from_a");
+    expect(results[1][0].val).toBe("from_b");
+    expect(results[2][0].val).toBe("from_a");
+    expect(results[3][0].val).toBe("from_b");
+
+    // Cleanup
+    await db`DROP TABLE IF EXISTS prepare_false_test_a`;
+    await db`DROP TABLE IF EXISTS prepare_false_test_b`;
+  });
+
+  test("parameterized query with multiple params", async () => {
+    await using db = new SQL(options);
+    const [{ sum }] = await db`SELECT (${10}::int + ${20}::int) AS sum`;
+    expect(sum).toBe(30);
+  });
+
+  test("query without params still works", async () => {
+    await using db = new SQL(options);
+    const [{ x }] = await db`SELECT 1 AS x`;
+    expect(x).toBe(1);
+  });
+
+  test("transactions with parameterized queries", async () => {
+    await using db = new SQL(options);
+
+    await db`CREATE TEMP TABLE IF NOT EXISTS tx_test (id int, val text)`;
+
+    await db.begin(async tx => {
+      await tx`INSERT INTO tx_test VALUES (${1}, ${"hello"})`;
+      await tx`INSERT INTO tx_test VALUES (${2}, ${"world"})`;
+    });
+
+    const rows = await db`SELECT * FROM tx_test ORDER BY id`;
+    expect(rows.length).toBe(2);
+    expect(rows[0].val).toBe("hello");
+    expect(rows[1].val).toBe("world");
+  });
+
+  test("concurrent parameterized queries with high concurrency", async () => {
+    await using db = new SQL({ ...options, max: 8 });
+
+    // Fire many concurrent queries to stress-test unnamed statement handling
+    const promises = [];
+    for (let i = 0; i < 50; i++) {
+      promises.push(db`SELECT ${i}::int AS x`.then(r => ({ expected: i, actual: r[0].x })));
+    }
+
+    const results = await Promise.all(promises);
+    for (const { expected, actual } of results) {
+      expect(actual).toBe(expected);
+    }
+  });
+});