From e3d69078f8af88368fdfbf8275af2464ab62a814 Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 22:29:24 +0300
Subject: [PATCH 1/7] Keep each session's statements prepared: Parse once, then
 Bind by name (#1)

A prepared statement's first execution on a connection parses it under a
name of the session's statement cache and describes it, in the same
exchange as its Bind and Execute; later executions send only Bind,
Execute and Sync. The cache is an LRU of statement_cache_capacity entries
(512, pgx's default; 0 turns it off, as does prepared_statements_cache=false),
a Close for each statement it gives up going with the next exchange.

A kept statement the server refuses to bind as out of date - 26000, 0A000
(cached plan must not change result type), class 42, or class 22 with
parameters - is given up and the query prepared anew once, when the error
left no transaction block aborted. DISCARD ALL and DEALLOCATE ALL empty
the cache, and DDL on the same connection gives up every kept statement,
whose parameter types were inferred against the schema before it.
Unprepared statements keep the unnamed Parse.
---
 pg/core.iyi                 | 131 +++++++++++++--
 pg/pq/connection.iyi        |  74 ++++++++-
 pg/pq/conninfo.iyi          |  27 ++-
 pg/pq/frame.iyi             |  52 ++++++
 pg/pq/statement_cache.iyi   | 138 ++++++++++++++++
 pg/statement_cache_test.iyi | 317 ++++++++++++++++++++++++++++++++++++
 6 files changed, 711 insertions(+), 28 deletions(-)
 create mode 100644 pg/pq/statement_cache.iyi
 create mode 100644 pg/statement_cache_test.iyi

diff --git a/pg/core.iyi b/pg/core.iyi
index f85291e..2ae9f06 100644
--- a/pg/core.iyi
+++ b/pg/core.iyi
@@ -19,7 +19,7 @@ import db
 import db/db/errors::*
 import db/db/value::{DBValue, ExecResult}
 import db/db/core::{Driver, ConnectionBuilder, ResultSet, Statement, Connection, ConnectionLost}
-import pg/pq/errors::{StreamError}
+import pg/pq/errors::{StreamError, PQError}
 import pg/pq/stream::{ByteSource}
 import pg/pq/frame::*
 import pg/pq/param::{Param}
@@ -349,10 +349,19 @@ pub class PGResultSet < ResultSet
 end
 
 pub class PGStatement < Statement
-  def initialize(connection : Connection, @conn : PQConnection, command : String)
+  # Whether the statement runs as a statement the session keeps prepared
+  # (`StatementCache`); otherwise it is parsed each time it runs, as the
+  # unnamed statement.
+  @cached : Bool
+
+  def initialize(connection : Connection, @conn : PQConnection, command : String, @cached : Bool = false)
     super(connection, command)
   end
 
+  def cached? : Bool
+    @cached
+  end
+
   protected def conn : PQConnection
     @conn
   end
@@ -383,13 +392,21 @@ pub class PGStatement < Statement
   # The most parameters a Bind message counts (an unsigned 16-bit number).
   MAX_PARAMETERS = 65535
 
-  # Parses, binds and executes the statement with the extended query
-  # protocol, and reads up to the description of its rows.
+  # Parses (unless the session keeps the statement prepared), binds and
+  # executes the statement with the extended query protocol, and reads up
+  # to the description of its rows.
   private def start_query(args : Array(DBValue)) : PGResultSet | DBError
     if args.size > MAX_PARAMETERS
       return DBError.new("#{args.size} parameters: a statement takes at most #{MAX_PARAMETERS}")
     end
     params = args.map { |arg| Param.encode(arg) }
+    return run_cached(params, retry: true) if @cached
+    run_unnamed(params)
+  end
+
+  # Parse, Bind, Describe portal, Execute and Sync: the unnamed statement,
+  # parsed and planned again on every execution.
+  private def run_unnamed(params : Array(Param)) : PGResultSet | DBError
     @conn.send_parse_message(command)
     @conn.send_bind_message(params)
     @conn.send_describe_portal_message
@@ -397,17 +414,99 @@ pub class PGStatement < Statement
     @conn.send_sync_message!
     @conn.expect_frame(Frame::ParseComplete)!
     @conn.expect_frame(Frame::BindComplete)!
-    frame = @conn.read!
-    fields = case frame
-             when Frame::RowDescription
-               frame.fields
-             when Frame::NoData
-               nil
-             else
-               return DBError.new("expected RowDescription or NoData, got #{frame}")
-             end
+    PGResultSet.new(self, @conn, row_fields!)
+  end
+
+  # Bind, Execute and Sync for the statement the session keeps prepared for
+  # this query. When it keeps none, the statement is prepared in the same
+  # exchange: Parse under a name of the cache's, and Describe statement,
+  # whose row description is kept with the name. A Close for each statement
+  # the cache gave up goes first.
+  #
+  # A kept statement the server refuses to bind as out of date (see
+  # `give_up_stale`) is given up, and the query runs again, prepared anew -
+  # once, while *retry*, and only when the error left the session outside a
+  # transaction block: inside one, the transaction is aborted and the error
+  # is the caller's.
+  private def run_cached(params : Array(Param), retry : Bool) : PGResultSet | DBError
+    cache = @conn.statement_cache
+    statement = cache.fetch?(command)
+    name = statement ? statement.name : cache.next_name
+    closing = cache.take_closing
+    closing.each { |old| @conn.send_close_statement_message(old) }
+    unless statement
+      @conn.send_parse_message(command, name)
+      @conn.send_describe_statement_message(name)
+    end
+    @conn.send_bind_message(params, name)
+    @conn.send_execute_message
+    @conn.send_sync_message!
+    closing.each do |_|
+      closed = @conn.expect_frame(Frame::CloseComplete)
+      return closed if closed.is_a?(DBError)
+    end
+
+    if statement
+      bound = @conn.expect_frame(Frame::BindComplete)
+      if bound.is_a?(DBError)
+        if give_up_stale(bound, params) && retry && @conn.transaction_status == 'I'
+          return run_cached(params, retry: false)
+        end
+        return bound
+      end
+      return PGResultSet.new(self, @conn, statement.fields)
+    end
+
+    @conn.expect_frame(Frame::ParseComplete)!
+    @conn.expect_frame(Frame::ParameterDescription)!
+    fields = row_fields
+    if fields.is_a?(DBError)
+      # prepared, but not kept: the server is asked to close it
+      cache.close_later(name)
+      return fields
+    end
+    cache.store(command, name, fields)
+    @conn.expect_frame(Frame::BindComplete)!
     PGResultSet.new(self, @conn, fields)
   end
+
+  # The fields of the RowDescription that comes next, nil for NoData.
+  private def row_fields : Array(Field) | Nil | DBError
+    frame = @conn.read!
+    case frame
+    when Frame::RowDescription
+      frame.fields
+    when Frame::NoData
+      nil
+    else
+      DBError.new("expected RowDescription or NoData, got #{frame}")
+    end
+  end
+
+  # Gives up the statement kept for this query when *error*, the server's
+  # answer to its Bind, may be the statement's being out of date rather
+  # than the query's fault; whether it did. Out of date: the statement is
+  # gone (26000, after a `DEALLOCATE`; every kept statement is given up,
+  # as a `DEALLOCATE ALL` or `DISCARD ALL` dropped them all), its rows
+  # changed type (0A000, "cached plan must not change result type", after
+  # an `ALTER TABLE`), analysing it again against the schema fails (class
+  # 42), or an argument does not convert to a parameter type inferred when
+  # it was prepared, which may no longer be the column's (class 22, for a
+  # statement with parameters). A query at fault fails the same way when
+  # prepared anew, and that error is answered.
+  private def give_up_stale(error : DBError, params : Array(Param)) : Bool
+    return false unless error.is_a?(PQError)
+    code = error.field_message(:code) || ""
+    cache = @conn.statement_cache
+    if code == "26000"
+      cache.evict_all
+    elsif code == "0A000" || code.starts_with?("42") || (code.starts_with?("22") && !params.empty?)
+      cache.evict(command)
+    else
+      return false
+    end
+    true
+  end
 end
 
 pub class PGConnection < Connection
@@ -426,10 +525,14 @@ pub class PGConnection < Connection
     PGConnection.new(options, pq)
   end
 
+  # A statement the session keeps prepared after its first execution, unless
+  # `prepared_statements_cache=false` or `statement_cache_capacity=0`.
   def build_prepared_statement(query : String) : Statement | DBError
-    PGStatement.new(self, @connection, query)
+    cached = prepared_statements_cache? && @connection.statement_cache.enabled?
+    PGStatement.new(self, @connection, query, cached: cached)
   end
 
+  # A statement parsed each time it runs, as the unnamed statement.
   def build_unprepared_statement(query : String) : Statement | DBError
     PGStatement.new(self, @connection, query)
   end
diff --git a/pg/pq/connection.iyi b/pg/pq/connection.iyi
index ae4d512..6fea06e 100644
--- a/pg/pq/connection.iyi
+++ b/pg/pq/connection.iyi
@@ -20,6 +20,7 @@ import pg/pq/stream::{Stream}
 import pg/pq/conninfo::{ConnInfo}
 import pg/pq/frame::*
 import pg/pq/param::{Param}
+import pg/pq/statement_cache::{StatementCache}
 import pg/pq/scram::{SaslContext, md5_password}
 import pg/pq/ssl::{tls_config, server_end_point}
 import pg/pq/pgpass::{LOG}
@@ -41,11 +42,20 @@ pub class PQConnection
   # When the attempt that opened this session runs out of `connect_timeout`,
   # on the monotonic clock in milliseconds; nil for no bound.
   @deadline_ms : Int64? = nil
+  # The statements this session keeps prepared on the server.
+  getter statement_cache : StatementCache
+  # The Parse messages this session has sent: one for every statement it
+  # prepared, named or unnamed.
+  getter parse_count : Int64 = 0_i64
+  # The transaction status of the last ReadyForQuery: `'I'` idle, `'T'` in
+  # a transaction block, `'E'` in a failed one.
+  getter transaction_status : Char = 'I'
 
   def initialize(@stream : Stream, @conninfo : ConnInfo)
     @server_parameters = {} of String => String
     @notice_handler = ->(notice : Notice) { nil }
     @notification_handler = ->(notification : Notification) { nil }
+    @statement_cache = StatementCache.new(@conninfo.statement_cache_capacity)
   end
 
   # The code an SSLRequest sends where a startup message sends its
@@ -341,6 +351,9 @@ pub class PQConnection
   end
 
   # `true` when *frame* was handled here and the caller reads the next one.
+  # A ReadyForQuery's transaction status is noted, and a command that
+  # dropped the session's prepared statements, or changed the schema,
+  # empties its cache; both are still the caller's to read.
   private def handle_async_frames(frame : Frame) : Bool | DBError
     case frame
     when Frame::ErrorResponse
@@ -370,6 +383,22 @@ pub class PQConnection
         # nothing to check
       end
       true
+    when Frame::ReadyForQuery
+      @transaction_status = frame.transaction_status
+      false
+    when Frame::CommandComplete
+      if frame.drops_statements?
+        @statement_cache.forget_all
+      elsif frame.changes_schema?
+        # A statement prepared before keeps the parameter types inferred
+        # then, and a value converts to an old one without an error: a
+        # `Time` bound to a `timestamp` parameter of a table recreated
+        # with a `timestamptz` column lost its zone. Prepared anew, the
+        # types are the new schema's. DDL from another session is not
+        # seen here; see the README.
+        @statement_cache.evict_all
+      end
+      false
     else
       false
     end
@@ -697,27 +726,33 @@ pub class PQConnection
     @stream.flush
   end
 
-  def send_parse_message(query : String) : Nil
+  # Parses *query* into the prepared statement *name*, the unnamed one when
+  # it is empty, leaving the parameters' types to the server.
+  def send_parse_message(query : String, name : String = "") : Nil
+    @parse_count += 1
     @stream.write_char('P')
-    @stream.write_i32(query.bytesize + 4 + 1 + 2 + 1)
-    @stream.write_byte(0_u8) # prepared statment name
+    @stream.write_i32(4 + name.bytesize + 1 + query.bytesize + 1 + 2)
+    @stream.write(name)
+    @stream.write_byte(0_u8)
     @stream.write(query)
     @stream.write_byte(0_u8)
     @stream.write_i16(0) # don't give any param types
   end
 
-  # *result_format* 1 asks for binary results, which every decoder reads; 0
-  # asks for text.
-  def send_bind_message(params : Array(Param), result_format : Int32 = 1) : Nil
+  # Binds *params* to the prepared statement *statement* (the unnamed one
+  # when it is empty) in the unnamed portal. *result_format* 1 asks for
+  # binary results, which every decoder reads; 0 asks for text.
+  def send_bind_message(params : Array(Param), statement : String = "", result_format : Int32 = 1) : Nil
     nparams = params.size
     total_size = params.reduce(0) do |acc, p|
       acc + 4 + (p.size == -1 ? 0 : p.size)
     end
 
     @stream.write_char('B')
-    @stream.write_i32(4 + 1 + 1 + 2 + (2 * nparams) + 2 + total_size + 2 + 2)
-    @stream.write_byte(0_u8)   # unnamed destination portal
-    @stream.write_byte(0_u8)   # unnamed prepared statment
+    @stream.write_i32(4 + 1 + statement.bytesize + 1 + 2 + (2 * nparams) + 2 + total_size + 2 + 2)
+    @stream.write_byte(0_u8) # unnamed destination portal
+    @stream.write(statement)
+    @stream.write_byte(0_u8)
     @stream.write_i16(nparams) # number of params format codes to follow
     params.each { |p| @stream.write_i16(p.format.to_i32) }
     @stream.write_i16(nparams) # number of params to follow
@@ -736,6 +771,27 @@ pub class PQConnection
     @stream.write_byte(0_u8)
   end
 
+  # Asks for the prepared statement *name*'s parameter types and the
+  # description of its rows: a ParameterDescription, then a RowDescription
+  # or NoData.
+  def send_describe_statement_message(name : String) : Nil
+    @stream.write_char('D')
+    @stream.write_i32(4 + 1 + name.bytesize + 1)
+    @stream.write_char('S')
+    @stream.write(name)
+    @stream.write_byte(0_u8)
+  end
+
+  # Closes the prepared statement *name*: a CloseComplete, even when the
+  # session has no statement by that name.
+  def send_close_statement_message(name : String) : Nil
+    @stream.write_char('C')
+    @stream.write_i32(4 + 1 + name.bytesize + 1)
+    @stream.write_char('S')
+    @stream.write(name)
+    @stream.write_byte(0_u8)
+  end
+
   def send_execute_message : Nil
     @stream.write_char('E')
     @stream.write_i32(4 + 1 + 4)
diff --git a/pg/pq/conninfo.iyi b/pg/pq/conninfo.iyi
index 7129031..53bfacd 100644
--- a/pg/pq/conninfo.iyi
+++ b/pg/pq/conninfo.iyi
@@ -44,6 +44,10 @@ pub struct ConnInfo
   # The `ssl_min_protocol_version`s std/tls speaks.
   TLS_VERSIONS = ["TLSv1.2", "TLSv1.3"]
 
+  # The prepared statements a session keeps unless `statement_cache_capacity`
+  # says otherwise: pgx's default.
+  STATEMENT_CACHE_CAPACITY = 512
+
   # The host. If it starts with a / it is the directory of a local Unix
   # socket.
   getter host : String
@@ -101,6 +105,12 @@ pub struct ConnInfo
   # statement_timeout=5000`. Optional.
   getter options : String?
 
+  # How many prepared statements a session keeps, by query text, so a query
+  # run again is not parsed again (`pg/pq/statement_cache`):
+  # `STATEMENT_CACHE_CAPACITY` unless given. 0 keeps none, and every
+  # statement is parsed each time it runs, as the unnamed statement.
+  getter statement_cache_capacity : Int32
+
   # A connection description from its parts; what is not given is taken from
   # the environment (`PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER`,
   # `PGPASSWORD`, `PGAPPNAME`) or the password file. The database defaults
@@ -125,6 +135,7 @@ pub struct ConnInfo
     @ssl_min_protocol_version = "TLSv1.2"
     @connect_timeout = connect_timeout ? ConnInfo.effective_timeout(connect_timeout) : nil
     @options = options.presence if options
+    @statement_cache_capacity = STATEMENT_CACHE_CAPACITY
   end
 
   # `connect_timeout` in milliseconds, nil for no bound.
@@ -211,9 +222,9 @@ pub struct ConnInfo
   # From a `URI`: its host (or the `host` parameter), port, path as the
   # database, user and password, and `application_name`, `replication`,
   # `sslmode`, `sslcert`, `sslkey`, `sslrootcert`, `channel_binding`,
-  # `ssl_min_protocol_version`, `connect_timeout`, `options` and
-  # `auth_methods` from its query; the `ENV_DEFAULTS` for those it does not
-  # give.
+  # `ssl_min_protocol_version`, `connect_timeout`, `options`,
+  # `statement_cache_capacity` and `auth_methods` from its query; the
+  # `ENV_DEFAULTS` for those it does not give.
   def self.from_uri(uri : URI) : ConnInfo | DBError
     params = Params.parse(uri.query || "")
     hostname = (uri.hostname || "").presence || params["host"]? || ""
@@ -233,8 +244,8 @@ pub struct ConnInfo
   # `"password"`, `"port"`, `"sslmode"`, `"sslcert"`, `"sslkey"`,
   # `"sslrootcert"`, `"channel_binding"`, `"ssl_min_protocol_version"`,
   # `"connect_timeout"`, `"options"`, `"application_name"`,
-  # `"replication"` and `"auth_methods"`; the `ENV_DEFAULTS` for those it
-  # does not give.
+  # `"replication"`, `"statement_cache_capacity"` and `"auth_methods"`; the
+  # `ENV_DEFAULTS` for those it does not give.
   def self.from_hash(params : Hash(String, String)) : ConnInfo | DBError
     port = parse_port(params["port"]?)! || env_port!
     info = ConnInfo.new(params["host"]?, params["dbname"]?, params["user"]?,
@@ -327,6 +338,12 @@ pub struct ConnInfo
       @connect_timeout = ConnInfo.effective_timeout(seconds)
     when "options"
       @options = value.presence
+    when "statement_cache_capacity"
+      capacity = value.strip.to_i?
+      if capacity.nil? || capacity < 0
+        return ArgumentError.new("invalid statement_cache_capacity #{value.inspect}: a number of statements, 0 for none")
+      end
+      @statement_cache_capacity = capacity
     when "auth_methods"
       methods = [] of String
       value.split(',').each do |method|
diff --git a/pg/pq/frame.iyi b/pg/pq/frame.iyi
index 8a4e7c7..283a11e 100644
--- a/pg/pq/frame.iyi
+++ b/pg/pq/frame.iyi
@@ -81,6 +81,8 @@ pub abstract class Frame
     when 'Z' then ReadyForQuery.new(bytes)
     when '1' then ParseComplete.new(bytes)
     when '2' then BindComplete.new(bytes)
+    when '3' then CloseComplete.new(bytes)
+    when 't' then ParameterDescription.new(bytes)
     when 'T' then RowDescription.new(bytes)
     when 'A' then NotificationResponse.new(bytes)
     when 'E' then ErrorResponse.new(bytes)
@@ -349,6 +351,35 @@ pub abstract class Frame
       last = command.split.last?
       (last && last.to_i64?) || 0_i64
     end
+
+    # Whether the command dropped every prepared statement of the session:
+    # `DISCARD ALL` or `DEALLOCATE ALL`. The tag's length is looked at
+    # first, so the other commands' tags are not made into strings.
+    def drops_statements? : Bool
+      size = @bytes.size
+      return false unless size == 12 || size == 15
+      tag = command
+      tag == "DISCARD ALL" || tag == "DEALLOCATE ALL"
+    end
+
+    # Whether the command was DDL - `CREATE ...`, `DROP ...`, `ALTER ...` -
+    # after which a prepared statement's parameter types, inferred when it
+    # was prepared, may no longer be what its text would give.
+    def changes_schema? : Bool
+      tag_starts_with?("CREATE ") || tag_starts_with?("DROP ") || tag_starts_with?("ALTER ")
+    end
+
+    private def tag_starts_with?(prefix : String) : Bool
+      size = prefix.bytesize
+      return false if @bytes.size <= size
+      bytes = prefix.to_unsafe
+      i = 0
+      while i < size
+        return false if @bytes[i] != bytes[i]
+        i += 1
+      end
+      true
+    end
   end
 
   class ParseComplete < Frame
@@ -361,6 +392,27 @@ pub abstract class Frame
     end
   end
 
+  class CloseComplete < Frame
+    def initialize(bytes : Bytes)
+    end
+  end
+
+  # The types of a prepared statement's parameters, as the server inferred
+  # them, by OID.
+  class ParameterDescription < Frame
+    getter type_oids : Array(Int32)
+
+    def initialize(bytes : Bytes)
+      pos, count = i16(0, bytes)
+      @type_oids = [] of Int32
+      # the count is unsigned on the wire: up to 65535 parameters
+      (count.to_i32 & 0xFFFF).times do
+        pos, oid = i32(pos, bytes)
+        @type_oids << oid
+      end
+    end
+  end
+
   class EmptyQueryResponse < Frame
     def initialize(bytes : Bytes)
     end
diff --git a/pg/pq/statement_cache.iyi b/pg/pq/statement_cache.iyi
new file mode 100644
index 0000000..e7c83a4
--- /dev/null
+++ b/pg/pq/statement_cache.iyi
@@ -0,0 +1,138 @@
+# The statements a session keeps prepared on the server, by query text, so
+# a query run again is bound and executed without being parsed and planned
+# again (iyilang/pg#1). pgx's statement cache, which crystal-pg has no
+# counterpart of.
+#
+# The parameters' types are not part of the key: a statement is parsed with
+# none given, so the server infers them from the query text alone, as it
+# does for the unnamed statement each execution parsed before.
+module pg/pq/statement_cache
+
+import pg/pq/frame::{Field}
+
+# One statement the server holds prepared under *name*, and the description
+# of its rows: nil for a statement that returns none.
+pub class CachedStatement
+  getter name : String
+  getter fields : Array(Field)?
+  # When it was last used, on its cache's clock.
+  property used : Int64
+
+  def initialize(@name : String, @fields : Array(Field)?, @used : Int64)
+  end
+end
+
+# At most `capacity` prepared statements, the least recently used one
+# making room for a new one. A capacity of 0 keeps none: every statement is
+# parsed each time it runs, as the unnamed statement.
+#
+# The server is not told here: the names of the statements it should close
+# wait in `closing` until the session's next exchange sends a Close for each.
+pub class StatementCache
+  getter capacity : Int32
+  @statements : Hash(String, CachedStatement)
+  # Counts uses, so the least recently used statement is the one whose last
+  # use is the smallest. A hit costs one write; the scan for the smallest
+  # happens only when a full cache takes a new statement.
+  @clock = 0_i64
+  # Numbers the names, which are never used twice in a session.
+  @serial = 0_i64
+  @closing : Array(String)
+
+  def initialize(@capacity : Int32)
+    @statements = {} of String => CachedStatement
+    @closing = [] of String
+  end
+
+  def enabled? : Bool
+    @capacity > 0
+  end
+
+  # How many statements are prepared.
+  def size : Int32
+    @statements.size
+  end
+
+  def includes?(query : String) : Bool
+    @statements.has_key?(query)
+  end
+
+  # The statement prepared for *query*, now its most recently used one; nil
+  # when there is none.
+  def fetch?(query : String) : CachedStatement?
+    if statement = @statements[query]?
+      @clock += 1
+      statement.used = @clock
+    end
+    statement
+  end
+
+  # A name for a statement about to be prepared. A full cache makes room
+  # first, by giving up its least recently used statement.
+  def next_name : String
+    evict_least_recent if @statements.size >= @capacity
+    @serial += 1
+    "iyi_pg_#{@serial}"
+  end
+
+  # *query* is prepared on the server under *name*, its rows described by
+  # *fields*.
+  def store(query : String, name : String, fields : Array(Field)?) : Nil
+    @clock += 1
+    @statements[query] = CachedStatement.new(name, fields, @clock)
+  end
+
+  # Gives up the statement prepared for *query*; the server is asked to
+  # close it.
+  def evict(query : String) : Nil
+    if statement = @statements.delete(query)
+      @closing << statement.name
+    end
+  end
+
+  # Gives up every statement; the server is asked to close each.
+  def evict_all : Nil
+    @statements.each_value { |statement| @closing << statement.name }
+    @statements.clear
+  end
+
+  # The server dropped every statement of the session (`DISCARD ALL`,
+  # `DEALLOCATE ALL`): there is nothing left to close either.
+  def forget_all : Nil
+    @statements.clear
+    @closing.clear
+  end
+
+  # A statement prepared under *name* that the cache does not hold, to be
+  # closed on the server.
+  def close_later(name : String) : Nil
+    @closing << name
+  end
+
+  # Whether names wait to be closed on the server.
+  def closing? : Bool
+    !@closing.empty?
+  end
+
+  # The names waiting to be closed, which the caller now sends a Close for;
+  # the cache forgets them.
+  def take_closing : Array(String)
+    names = @closing
+    @closing = [] of String
+    names
+  end
+
+  private def evict_least_recent : Nil
+    oldest = nil.as(String?)
+    oldest_use = 0_i64
+    @statements.each do |query, statement|
+      if oldest.nil? || statement.used < oldest_use
+        oldest = query
+        oldest_use = statement.used
+      end
+    end
+    if query = oldest
+      evict(query)
+    end
+  end
+end
diff --git a/pg/statement_cache_test.iyi b/pg/statement_cache_test.iyi
new file mode 100644
index 0000000..be66bc7
--- /dev/null
+++ b/pg/statement_cache_test.iyi
@@ -0,0 +1,317 @@
+# The session's prepared statements (iyilang/pg#1): a query is parsed once
+# per connection, then bound and executed by name. The Parse messages a
+# session sends are counted by `PQConnection#parse_count`; what the server
+# holds is read from `pg_prepared_statements` with unprepared queries, which
+# use the unnamed statement and leave the cache alone.
+module pg/statement_cache_test
+
+import std/uri::{URI}
+import std/time::{Time}
+import db
+import db/db/errors::*
+import db/db/core::{Database, Connection}
+import db/db/spec::{describe, it, expect_error}
+import pg/pq/errors::{PQError}
+import pg/pq/conninfo::{ConnInfo}
+import pg/pq/statement_cache::{StatementCache}
+import pg/core::{PGConnection}
+import pg/testing/spec_helper::*
+
+# DB_URL with *params* added to its query.
+def url_with(params : String) : String
+  uri = URI.parse(DB_URL)
+  query = uri.query
+  uri.query = query.nil? || query.empty? ? params : "#{query}&#{params}"
+  uri.to_s
+end
+
+def with_connection_at(url : String, & : PGConnection -> Nil) : Nil
+  ::Db.connect(url) { |conn| yield conn.as(PGConnection) }.or_panic
+end
+
+def parses(conn : PGConnection) : Int64
+  conn.connection.parse_count
+end
+
+# The statements the server holds prepared for the session, by text.
+def server_statements(conn : PGConnection) : Array(String)
+  texts = [] of String
+  conn.unprepared.query("select statement from pg_prepared_statements order by statement") do |rs|
+    rs.each { texts << rs.read(String).or_panic; nil }.or_panic
+    nil
+  end.or_panic
+  texts
+end
+
+def code_of(result : V) : String forall V
+  if result.is_a?(PQError)
+    result.field_message(:code) || ""
+  else
+    "not a PQError: #{result.inspect}"
+  end
+end
+
+Q1 = "select $1::int4 + 1"
+Q2 = "select $1::int4 + 2"
+Q3 = "select $1::int4 + 3"
+
+describe "PG statement cache" do
+  it "parses a query once per connection, then binds it by name" do
+    with_connection do |conn|
+      before = parses(conn)
+      assert conn.scalar(Q1, 1).or_panic == 2
+      assert parses(conn) == before + 1
+      3.times { |i| assert conn.scalar(Q1, i).or_panic == i + 1 }
+      assert parses(conn) == before + 1, "#{parses(conn) - before} parses"
+      assert conn.connection.statement_cache.includes?(Q1)
+      assert server_statements(conn).includes?(Q1)
+    end
+  end
+
+  it "keeps the description of rows and of no rows" do
+    with_connection do |conn|
+      conn.exec("create temporary table cache_rows (a int4, b text)").or_panic
+      2.times do |i|
+        conn.exec("insert into cache_rows values ($1, $2)", i, "row #{i}").or_panic
+      end
+      2.times do
+        conn.query("select a, b from cache_rows order by a") do |rs|
+          assert rs.column_count == 2
+          assert rs.column_name(1) == "b"
+          got = [] of String
+          rs.each { got << "#{rs.read(Int32).or_panic} #{rs.read(String).or_panic}"; nil }.or_panic
+          assert got == ["0 row 0", "1 row 1"], got.inspect
+          nil
+        end.or_panic
+      end
+      result = conn.exec("update cache_rows set b = $1 where a = $2", "x", 1).or_panic
+      assert result.rows_affected == 1
+      result = conn.exec("update cache_rows set b = $1 where a = $2", "y", 7).or_panic
+      assert result.rows_affected == 0
+    end
+  end
+
+  it "evicts the least recently used statement, closing it on the server" do
+    with_connection_at(url_with("statement_cache_capacity=2")) do |conn|
+      cache = conn.connection.statement_cache
+      assert cache.capacity == 2
+      conn.scalar(Q1, 1).or_panic
+      conn.scalar(Q2, 1).or_panic
+      conn.scalar(Q1, 1).or_panic # Q1 is now the more recently used
+      before = parses(conn)
+      conn.scalar(Q3, 1).or_panic # makes room by giving up Q2
+      assert parses(conn) == before + 1
+      assert cache.size == 2
+      assert cache.includes?(Q1) && cache.includes?(Q3) && !cache.includes?(Q2)
+      assert server_statements(conn) == [Q1, Q3], server_statements(conn).inspect
+      # Q1 was kept: no Parse; Q2 was not: parsed again
+      before = parses(conn)
+      assert conn.scalar(Q1, 5).or_panic == 6
+      assert parses(conn) == before
+      assert conn.scalar(Q2, 5).or_panic == 7
+      assert parses(conn) == before + 1
+      assert server_statements(conn).size == 2
+    end
+  end
+
+  it "prepares a statement again when ALTER TABLE elsewhere changed its rows' type" do
+    PG_DB.exec("drop table if exists cache_alter").or_panic
+    PG_DB.exec("create table cache_alter (a int4)").or_panic
+    PG_DB.exec("insert into cache_alter values (1)").or_panic
+    with_connection do |conn|
+      assert conn.query_one("select * from cache_alter", as: Int32).or_panic == 1
+      with_connection { |other| other.exec("alter table cache_alter alter column a type text").or_panic }
+      # the kept statement fails to bind with 0A000, "cached plan must not
+      # change result type"; it is given up and the query prepared anew
+      before = parses(conn)
+      assert conn.query_one("select * from cache_alter", as: String).or_panic == "1"
+      assert parses(conn) == before + 1
+      with_connection { |other| other.exec("alter table cache_alter add column b int4").or_panic }
+      assert conn.query("select * from cache_alter") { |rs| rs.column_count }.or_panic == 2
+      assert server_statements(conn).count { |text| text == "select * from cache_alter" } == 1
+    end
+    PG_DB.exec("drop table cache_alter").or_panic
+  end
+
+  it "prepares a statement again when a column's type changed elsewhere under its parameter" do
+    PG_DB.exec("drop table if exists cache_param").or_panic
+    PG_DB.exec("create table cache_param (a int4)").or_panic
+    with_connection do |conn|
+      conn.exec("insert into cache_param values ($1)", 7).or_panic
+      with_connection { |other| other.exec("alter table cache_param alter column a type text").or_panic }
+      # $1 was inferred as int4; "seven" does not convert to it (22P02)
+      conn.exec("insert into cache_param values ($1)", "seven").or_panic
+      assert conn.query_one("select count(*) from cache_param where a = $1", "seven", as: Int64).or_panic == 1_i64
+    end
+    PG_DB.exec("drop table cache_param").or_panic
+  end
+
+  it "prepares its statements anew after DDL on the same connection" do
+    with_connection do |conn|
+      conn.exec("set time zone 'Europe/Istanbul'").or_panic
+      at = Time.utc(2019, 8, 13, 16, 30)
+      conn.exec("create temporary table cache_ddl (v timestamp)").or_panic
+      conn.exec("insert into cache_ddl values ($1)", at).or_panic
+      conn.exec("drop table cache_ddl").or_panic
+      assert conn.connection.statement_cache.size == 0
+      conn.exec("create temporary table cache_ddl (v timestamptz)").or_panic
+      # kept, $1 would still be a timestamp, and 16:30 read in Istanbul
+      conn.exec("insert into cache_ddl values ($1)", at).or_panic
+      got = conn.query_one("select v from cache_ddl", as: Time).or_panic
+      assert got == at, got.inspect
+    end
+  end
+
+  it "answers the error inside a transaction, and prepares anew after it" do
+    PG_DB.exec("drop table if exists cache_tx").or_panic
+    PG_DB.exec("create table cache_tx (a int4)").or_panic
+    PG_DB.exec("insert into cache_tx values (1)").or_panic
+    with_connection do |conn|
+      assert conn.query_one("select * from cache_tx", as: Int32).or_panic == 1
+      with_connection { |other| other.exec("alter table cache_tx alter column a type int8").or_panic }
+      tx = conn.begin_transaction.or_panic
+      # no retry: the transaction is aborted
+      assert code_of(conn.query_one("select * from cache_tx", as: Int64)) == "0A000"
+      tx.rollback.or_panic
+      assert conn.query_one("select * from cache_tx", as: Int64).or_panic == 1_i64
+    end
+    PG_DB.exec("drop table cache_tx").or_panic
+  end
+
+  it "keeps its statements through an error in a transaction" do
+    with_connection do |conn|
+      assert conn.scalar(Q1, 1).or_panic == 2
+      tx = conn.begin_transaction.or_panic
+      before = parses(conn)
+      assert conn.scalar(Q1, 2).or_panic == 3
+      assert code_of(conn.exec("select 1/0")) == "22012"
+      # bound in the aborted transaction: refused, and not taken as stale
+      assert code_of(conn.scalar(Q1, 3)) == "25P02"
+      # only `select 1/0` was parsed
+      assert parses(conn) == before + 1, "#{parses(conn) - before} parses"
+      tx.rollback.or_panic
+      before = parses(conn)
+      assert conn.scalar(Q1, 4).or_panic == 5
+      assert parses(conn) == before
+      assert conn.connection.statement_cache.includes?(Q1)
+    end
+  end
+
+  it "prepares again after DEALLOCATE ALL, DISCARD ALL or a DEALLOCATE of one" do
+    with_connection do |conn|
+      conn.scalar(Q1, 1).or_panic
+      conn.scalar(Q2, 1).or_panic
+      cache = conn.connection.statement_cache
+      conn.exec("deallocate all").or_panic
+      assert cache.size == 0
+      assert server_statements(conn).empty?
+      assert conn.scalar(Q1, 1).or_panic == 2
+
+      conn.unprepared.exec("discard all").or_panic
+      assert cache.size == 0
+      assert conn.scalar(Q1, 1).or_panic == 2
+
+      # a statement dropped behind the cache's back: 26000 when bound, and
+      # the query prepared anew
+      name = server_name(conn, Q1)
+      conn.unprepared.exec("deallocate #{name}").or_panic
+      before = parses(conn)
+      assert conn.scalar(Q1, 2).or_panic == 3
+      assert parses(conn) == before + 1
+      assert server_statements(conn) == [Q1], server_statements(conn).inspect
+    end
+  end
+
+  it "is the connection's, through checkouts by other tasks" do
+    with_db_at(url_with("max_pool_size=1&initial_pool_size=1")) do |db|
+      pid = 0
+      before = 0_i64
+      db.using_connection do |conn|
+        pg = conn.as(PGConnection)
+        pid = pg.backend_pid
+        assert pg.scalar(Q1, 1).or_panic == 2
+        before = parses(pg)
+        nil
+      end.or_panic
+      group do |g|
+        3.times do |i|
+          g.spawn do
+            assert db.scalar(Q1, i).or_panic == i + 1
+            nil
+          end
+        end
+      end
+      db.using_connection do |conn|
+        pg = conn.as(PGConnection)
+        assert pg.backend_pid == pid
+        assert parses(pg) == before, "#{parses(pg) - before} parses"
+        nil
+      end.or_panic
+    end
+  end
+
+  it "is off with statement_cache_capacity=0, prepared_statements_cache=false, or unprepared" do
+    ["statement_cache_capacity=0", "prepared_statements_cache=false", "prepared_statements=false"].each do |off|
+      with_connection_at(url_with(off)) do |conn|
+        before = parses(conn)
+        3.times { |i| assert conn.scalar(Q1, i).or_panic == i + 1 }
+        assert parses(conn) == before + 3, "#{off}: #{parses(conn) - before} parses"
+        assert conn.connection.statement_cache.size == 0
+        assert server_statements(conn).empty?, off
+      end
+    end
+    with_connection do |conn|
+      before = parses(conn)
+      2.times { assert conn.unprepared.scalar("select 1").or_panic == 1 }
+      assert parses(conn) == before + 2
+      assert conn.connection.statement_cache.size == 0
+    end
+  end
+
+  it "reads statement_cache_capacity from the URL, and refuses a bad one" do
+    assert ConnInfo.from_conninfo_string("postgres://u@h/d").or_panic.statement_cache_capacity == ConnInfo::STATEMENT_CACHE_CAPACITY
+    assert ConnInfo.from_conninfo_string("postgres://u@h/d?statement_cache_capacity=7").or_panic.statement_cache_capacity == 7
+    assert ConnInfo.from_conninfo_string("host=h statement_cache_capacity=0").or_panic.statement_cache_capacity == 0
+    expect_error(ConnInfo.from_conninfo_string("postgres://u@h/d?statement_cache_capacity=-1"), ArgumentError)
+    expect_error(ConnInfo.from_conninfo_string("postgres://u@h/d?statement_cache_capacity=many"), ArgumentError)
+    nil
+  end
+end
+
+describe "PG StatementCache" do
+  it "names statements once each, and closes what it gives up" do
+    cache = StatementCache.new(2)
+    a = cache.next_name
+    cache.store("a", a, nil)
+    b = cache.next_name
+    cache.store("b", b, nil)
+    assert a != b
+    assert !cache.closing?
+    cache.fetch?("a")
+    c = cache.next_name # full: b, the least recently used, goes
+    assert c != a && c != b
+    cache.store("c", c, nil)
+    assert cache.take_closing == [b]
+    assert cache.take_closing.empty?
+    cache.evict("a")
+    cache.evict("missing")
+    assert cache.take_closing == [a]
+    cache.evict_all
+    assert cache.take_closing == [c]
+    assert cache.size == 0
+    cache.store("d", cache.next_name, nil)
+    cache.close_later("orphan")
+    cache.forget_all
+    assert cache.size == 0 && !cache.closing?
+    assert !StatementCache.new(0).enabled?
+  end
+end
+
+# The name the server holds *query* prepared under.
+def server_name(conn : PGConnection, query : String) : String
+  conn.unprepared.query_one("select name from pg_prepared_statements where statement = $1", query, as: String).or_panic
+end
+
+def with_db_at(url : String, & : Database -> Nil) : Nil
+  ::Db.open(url) { |db| yield db }.or_panic
+end

From 15c89d26eb462ae3bf9bc36b8e3b840fb84294d8 Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 22:33:59 +0300
Subject: [PATCH 2/7] Keep ConnInfo's tables and RangeValue's QUOTED in Lists

They are only read, from whichever worker a connection is opened or a
range is bound on; as Arrays each was a #148 reach warning in every
parallel block that opens a connection.
---
 pg/pq/conninfo.iyi | 19 +++++++++++--------
 pg/range.iyi       |  6 ++++--
 2 files changed, 15 insertions(+), 10 deletions(-)

diff --git a/pg/pq/conninfo.iyi b/pg/pq/conninfo.iyi
index 53bfacd..2958bf8 100644
--- a/pg/pq/conninfo.iyi
+++ b/pg/pq/conninfo.iyi
@@ -9,28 +9,31 @@ import std/path::{Path}
 import std/text
 import std/uri::{URI, Params}
 import std/user::{User}
+import std/list::{List}
 import db/db/errors::{DBError, ArgumentError}
 import pg/pq/pgpass::{PgPass, LOG}
 
 pub struct ConnInfo
-  SOCKET_SEARCH = ["/run/postgresql", "/tmp", "/var/run/postgresql"]
+  # These tables are `List`s, which any thread may read (SPEC.md III.4.4):
+  # a connection is opened from whichever worker its task runs on.
+  SOCKET_SEARCH = List(String).new(["/run/postgresql", "/tmp", "/var/run/postgresql"])
 
-  SUPPORTED_AUTH_METHODS = ["cleartext", "md5", "scram-sha-256", "scram-sha-256-plus"]
+  SUPPORTED_AUTH_METHODS = List(String).new(["cleartext", "md5", "scram-sha-256", "scram-sha-256-plus"])
 
-  SSL_MODES = [:disable, :allow, :prefer, :require, :"verify-ca", :"verify-full"]
+  SSL_MODES = List(Symbol).new([:disable, :allow, :prefer, :require, :"verify-ca", :"verify-full"])
 
   # The keys a URI's query or a `key=value` string may hold besides the ones
   # `handle_param` reads: the connection's own, and iyi-db's pool options.
   # Any other is reported, not silently dropped.
-  KNOWN_KEYS = [
+  KNOWN_KEYS = List(String).new([
     "host", "port", "dbname", "user", "password", "application_name", "replication",
     "initial_pool_size", "max_pool_size", "max_idle_pool_size", "checkout_timeout",
     "retry_attempts", "retry_delay", "prepared_statements", "prepared_statements_cache",
-  ]
+  ])
 
   # The parameters a `PG*` variable gives when the URL or string does not,
   # each with its variable, as libpq has them.
-  ENV_DEFAULTS = [
+  ENV_DEFAULTS = List(Tuple(String, String)).new([
     {"sslmode", "PGSSLMODE"},
     {"sslrootcert", "PGSSLROOTCERT"},
     {"sslcert", "PGSSLCERT"},
@@ -39,10 +42,10 @@ pub struct ConnInfo
     {"ssl_min_protocol_version", "PGSSLMINPROTOCOLVERSION"},
     {"connect_timeout", "PGCONNECT_TIMEOUT"},
     {"options", "PGOPTIONS"},
-  ]
+  ])
 
   # The `ssl_min_protocol_version`s std/tls speaks.
-  TLS_VERSIONS = ["TLSv1.2", "TLSv1.3"]
+  TLS_VERSIONS = List(String).new(["TLSv1.2", "TLSv1.3"])
 
   # The prepared statements a session keeps unless `statement_cache_capacity`
   # says otherwise: pgx's default.
diff --git a/pg/range.iyi b/pg/range.iyi
index 7de722f..8053e70 100644
--- a/pg/range.iyi
+++ b/pg/range.iyi
@@ -11,6 +11,7 @@ module pg/range
 
 import std/int
 import std/text
+import std/list::{List}
 import db/db/errors::{DBError}
 import db/db/value::{DBValue, CustomValue}
 import pg/pq/stream::{ByteSource}
@@ -87,8 +88,9 @@ pub class RangeValue < CustomValue
     "PG::RangeValue(#{to_s})"
   end
 
-  # The characters that make a printed bound quoted, besides whitespace.
-  QUOTED = ['"', '\\', ',', '(', ')', '[', ']']
+  # The characters that make a printed bound quoted, besides whitespace: a
+  # `List`, which any thread may read.
+  QUOTED = List(Char).new(['"', '\\', ',', '(', ')', '[', ']'])
 
   # PostgreSQL's `range_bound_escape`.
   private def write_bound(io : String::Builder, bound : DBValue) : Nil

From 7eedd112412e3d1e9e823c25a00bd4215246dae2 Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 22:33:59 +0300
Subject: [PATCH 3/7] Log through a Logger behind a Mutex

std's Logger is not Share: level=, backend= and for change it in place,
and a connection logs from whichever worker its task runs on. SharedLog
holds Log.for(source) in a Mutex and builds a message only when the
logger takes its severity. The tests set the backend on Log.for("pg"),
the same logger.
---
 pg/pq/conninfo_test.iyi | 10 +++++-----
 pg/pq/pgpass.iyi        | 42 +++++++++++++++++++++++++++++++++++++++--
 pg/pq/pgpass_test.iyi   |  8 ++++----
 pg/replication.iyi      | 12 ++++++------
 4 files changed, 55 insertions(+), 17 deletions(-)

diff --git a/pg/pq/conninfo_test.iyi b/pg/pq/conninfo_test.iyi
index 6d3ae0b..e9ba80a 100644
--- a/pg/pq/conninfo_test.iyi
+++ b/pg/pq/conninfo_test.iyi
@@ -2,11 +2,10 @@
 module pg/pq/conninfo_test
 
 import std/env
-import std/log::{MemoryBackend, Severity}
+import std/log::{Log, MemoryBackend, Severity}
 import db/db/errors::*
 import db/db/spec::{describe, it, expect_error}
 import pg/pq/conninfo::{ConnInfo}
-import pg/pq/pgpass::{LOG}
 import pg/testing/spec_helper::{env_var_bubble}
 
 def assert_default_params(ci : ConnInfo) : Nil
@@ -159,10 +158,11 @@ describe "PQ::ConnInfo.from_conninfo_string" do
 
   it "reports a parameter it ignores, and not iyi-db's pool options" do
     backend = MemoryBackend.new
-    previous = LOG.own_backend
-    LOG.backend = backend
+    logger = Log.for("pg")
+    previous = logger.own_backend
+    logger.backend = backend
     parse("postgres://host/db?sslmdoe=verify-full&max_pool_size=2&application_name=x&connect_timeout=5&options=-c%20a%3Db")
-    LOG.backend = previous
+    logger.backend = previous
     warnings = backend.entries.select { |e| e.severity >= Severity::Warn }.map(&.message)
     assert warnings == ["Ignoring the connection parameter sslmdoe, which this driver does not support"], warnings.inspect
   end
diff --git a/pg/pq/pgpass.iyi b/pg/pq/pgpass.iyi
index b0d49b9..a3cad5d 100644
--- a/pg/pq/pgpass.iyi
+++ b/pg/pq/pgpass.iyi
@@ -8,10 +8,48 @@ import std/env
 import std/file
 import std/path::{Path}
 import std/text
-import std/log::{Log, Logger}
+import std/log::{Log, Logger, Severity}
+import std/mutex::{Mutex}
+
+# A std `Logger` behind a lock, so a task on any worker may log through it:
+# std's `Logger` is not `Share` - `level=`, `backend=` and `for` change it
+# in place - and a connection logs from whichever worker its task runs on.
+# The message is built only when the logger takes its severity, as the
+# `Logger`'s block forms build it.
+#
+# The logger itself is `Log.for(source)`'s, which a program configures as
+# any std logger: `Log.for("pg").level = Severity::Debug`, at startup.
+pub class SharedLog
+  @logger : Mutex(Logger)
+
+  def initialize(source : String)
+    @logger = Mutex.new { Log.for(source) }
+  end
+
+  def debug(& : -> String) : Nil
+    write(Severity::Debug) { yield }
+  end
+
+  def warn(& : -> String) : Nil
+    write(Severity::Warn) { yield }
+  end
+
+  def error(& : -> String) : Nil
+    write(Severity::Error) { yield }
+  end
+
+  # A task cancelled while it waits for the lock writes nothing.
+  private def write(severity : Severity, & : -> String) : Nil
+    wanted = @logger.get { |log| severity >= log.level && severity != Severity::None }
+    return unless wanted == true
+    message = yield
+    @logger.lock { |log| log.log(severity, message) }
+    nil
+  end
+end
 
 # The logger for this library, crystal-pg's `PQ::Log`; its source is "pg".
-pub LOG = Log.for("pg")
+pub LOG = SharedLog.new("pg")
 
 pub class PgPass
   # The password the file holds for *host*, *port*, *db* and *user*, or nil.
diff --git a/pg/pq/pgpass_test.iyi b/pg/pq/pgpass_test.iyi
index fcb8708..f0bca56 100644
--- a/pg/pq/pgpass_test.iyi
+++ b/pg/pq/pgpass_test.iyi
@@ -6,7 +6,6 @@ import std/file
 import std/log::{Log, MemoryBackend, Severity}
 import db/db/spec::{describe, it}
 import pg/pq/conninfo::{ConnInfo}
-import pg/pq/pgpass::{LOG}
 import pg/testing/spec_helper::{env_var_bubble}
 
 def create_empty_pgpass_file(& : String -> Nil) : Nil
@@ -34,9 +33,10 @@ end
 # the block runs.
 def capture_pg_log(& : -> Nil) : MemoryBackend
   backend = MemoryBackend.new
-  previous = LOG.own_backend
-  LOG.backend = backend
-  defer LOG.backend = previous
+  logger = Log.for("pg")
+  previous = logger.own_backend
+  logger.backend = backend
+  defer logger.backend = previous
   yield
   backend
 end
diff --git a/pg/replication.iyi b/pg/replication.iyi
index baa9c30..5c60534 100644
--- a/pg/replication.iyi
+++ b/pg/replication.iyi
@@ -19,15 +19,15 @@ module pg/replication
 import std/int
 import std/uri::{URI}
 import std/time::{Span}
-import std/log::{Log}
 import db
 import db/db/errors::{DBError}
 import pg/pq/errors::{StreamError}
 import pg/pq/stream::{Stream}
+import pg/pq/pgpass::{SharedLog}
 import pg/core::{PGConnection}
 import pg/replication/messages::*
 
-LOG = Log.for("pg.replication")
+LOG = SharedLog.new("pg.replication")
 
 # What a `Connection` hands the stream to. crystal-pg's `module Handler`,
 # included by the user's class; here an abstract class to inherit, with a
@@ -211,11 +211,11 @@ pub class Connection
       break if closed?
       case result
       when StreamError
-        LOG.warn("Replication connection lost; reconnecting: #{result.message}")
+        LOG.warn { "Replication connection lost; reconnecting: #{result.message}" }
       when DBError
-        LOG.error("Unexpected error in replication loop; reconnecting: #{result.message}")
+        LOG.error { "Unexpected error in replication loop; reconnecting: #{result.message}" }
       else
-        LOG.warn("Replication connection lost; reconnecting")
+        LOG.warn { "Replication connection lost; reconnecting" }
       end
 
       # The connection dropped. Re-establish it, backing off if the server
@@ -228,7 +228,7 @@ pub class Connection
         if reconnected.is_a?(DBError)
           backoff = backoff + backoff
           backoff = MAX_RECONNECT_BACKOFF if backoff > MAX_RECONNECT_BACKOFF
-          LOG.error("Failed to reconnect replication; retrying in #{backoff.total_seconds}s: #{reconnected.message}")
+          LOG.error { "Failed to reconnect replication; retrying in #{backoff.total_seconds}s: #{reconnected.message}" }
         else
           backoff = BASE_RECONNECT_BACKOFF
           break

From 4eee6e99779d86b06e83282054cb0fb644439ecd Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 22:33:59 +0300
Subject: [PATCH 4/7] Read decoders from frozen tables

DECODERS was a Hash every column read looked up and register_decoder
wrote, from any thread. Registrations now go into a registry behind a
Mutex, counted by an Atomic, which builds a DecoderTable of them - the
OIDs sorted in a List, their decoders in another - when first asked after
one. A connection keeps the table it was given and asks again only after a
registration, so a column is read without a lock; Decoders.from_oid, which
records, ranges and typed arrays use for what they hold, takes the lock.
The decoders that held their OIDs in an Array hold a List, so a decoder
is Share.
---
 pg/core.iyi          |   2 +-
 pg/decoder.iyi       | 121 ++++++++++++++++++++++++++++++++++++++++---
 pg/pq/connection.iyi |  14 +++++
 pg/range.iyi         |  18 +++++--
 pg/record.iyi        |  10 +++-
 5 files changed, 151 insertions(+), 14 deletions(-)

diff --git a/pg/core.iyi b/pg/core.iyi
index 2ae9f06..4fb99fa 100644
--- a/pg/core.iyi
+++ b/pg/core.iyi
@@ -300,7 +300,7 @@ pub class PGResultSet < ResultSet
   end
 
   private def decoder(index : Int32) : Decoder
-    Decoders.from_oid(oid(index))
+    @conn.decoders.from_oid(oid(index))
   end
 
   private def oid(index : Int32) : Int32
diff --git a/pg/decoder.iyi b/pg/decoder.iyi
index 4476463..37b15e0 100644
--- a/pg/decoder.iyi
+++ b/pg/decoder.iyi
@@ -22,6 +22,8 @@ import pg/geo::{Point, Line, Circle, LineSegment, Box, Path, Polygon}
 import pg/time_tz::{TimeTz}
 import pg/inet::{Inet}
 import std/netip::{IP, Prefix}
+import std/list::{List}
+import std/mutex::{Mutex}
 
 # A `uuid` column. `DBValue` is a closed set and `std/uuid`'s `UUID` is not
 # in it, so the value a plain `read` answers wraps one; `read(UUID)` answers
@@ -700,10 +702,15 @@ end
 # with `nil` for NULL. An array whose lower bound is above 1 is padded with
 # `nil`s in front, as crystal-pg reads it.
 pub class ArrayDecoder < Decoder
-  getter oids : Array(Int32)
+  @oids : List(Int32)
   getter element : Decoder
 
-  def initialize(@oids : Array(Int32), @element : Decoder)
+  def initialize(oids : Array(Int32), @element : Decoder)
+    @oids = List(Int32).new(oids)
+  end
+
+  def oids : Array(Int32)
+    @oids.to_a
   end
 
   def type_name : String
@@ -768,21 +775,121 @@ pub class ArrayDecoder < Decoder
   end
 end
 
-DECODERS        = {} of Int32 => Decoder
 DEFAULT_DECODER = ByteaDecoder.new
 
+# The decoders by OID as some registration left them: the OIDs sorted, in a
+# `List`, and their decoders in another, never changed after they are
+# built, so any thread may read one (SPEC.md III.4.4). *generation* is the
+# count of registrations it holds.
+pub class DecoderTable
+  getter generation : Int64
+  @oids : List(Int32)
+  @decoders : List(Decoder)
+
+  def initialize(by_oid : Hash(Int32, Decoder), @generation : Int64)
+    oids = by_oid.keys
+    oids.sort_in_place
+    @oids = List(Int32).new(oids)
+    @decoders = List(Decoder).new(oids.map { |oid| by_oid[oid] })
+  end
+
+  # The decoder for *oid*; `Bytes` for an OID nobody registered.
+  def from_oid(oid : Int32) : Decoder
+    find(oid) || DEFAULT_DECODER
+  end
+
+  # The decoder registered for *oid*, by binary search; nil when none is.
+  def find(oid : Int32) : Decoder?
+    low = 0
+    high = @oids.size - 1
+    while low <= high
+      middle = (low + high) // 2
+      found = @oids[middle]
+      return @decoders[middle] if found == oid
+      if found < oid
+        low = middle + 1
+      else
+        high = middle - 1
+      end
+    end
+    nil
+  end
+
+  # Whether every registration *generation* counts is in this table.
+  def current?(generation : Int64) : Bool
+    @generation >= generation
+  end
+end
+
+# What `Decoders` keeps behind its lock: every registration, and the table
+# of them, built when it is first asked for after one.
+class DecoderRegistry
+  @by_oid : Hash(Int32, Decoder)
+  @count : Int64
+  @table : DecoderTable?
+
+  def initialize
+    @by_oid = {} of Int32 => Decoder
+    @count = 0_i64
+    @table = nil
+  end
+
+  def register(oids : List(Int32), decoder : Decoder) : Nil
+    oids.each { |oid| @by_oid[oid] = decoder }
+    @count += 1
+    @table = nil
+  end
+
+  def table : DecoderTable
+    if table = @table
+      return table
+    end
+    table = DecoderTable.new(@by_oid, @count)
+    @table = table
+    table
+  end
+end
+
 # The decoder registry, and array reading by element type.
+#
+# Registrations go into a `DecoderRegistry` behind a lock, from any thread
+# at any time. A connection reads its columns through a `DecoderTable` it
+# holds (`PQConnection#decoders`), taken from the registry again only after
+# a registration, so a column is read without a lock; `from_oid` takes the
+# lock.
 pub struct Decoders
+  @@registry : Mutex(DecoderRegistry) = Mutex.new { DecoderRegistry.new }
+  # Counts registrations, after each is in the registry: a table that
+  # holds fewer is out of date.
+  @@generation : Atomic(Int64) = Atomic(Int64).new(0_i64)
+
   # The decoder for *oid*; `Bytes` for an OID nobody registered.
   def self.from_oid(oid : Int32) : Decoder
-    DECODERS[oid]? || DEFAULT_DECODER
+    table.from_oid(oid)
   end
 
-  # Registers *decoder* for each of its OIDs.
+  # Registers *decoder* for each of its OIDs, in place of any registered
+  # before for them. A decoder is read by every thread, so it is `Share`:
+  # its fields are set in `initialize` only, and are `Share` themselves.
   def self.register_decoder(decoder : Decoder) : Nil
-    decoder.oids.each do |oid|
-      DECODERS[oid] = decoder
+    oids = List(Int32).new(decoder.oids)
+    # A task cancelled while it waits for the lock still registers.
+    while @@registry.lock { |registry| registry.register(oids, decoder) }.is_a?(Cancelled)
+    end
+    @@generation.add(1_i64)
+    nil
+  end
+
+  # The count of registrations so far.
+  def self.generation : Int64
+    @@generation.get
+  end
+
+  # The table of every registration so far.
+  def self.table : DecoderTable
+    while (table = @@registry.get { |registry| registry.table }).is_a?(Cancelled)
     end
+    table
   end
 
   def self.decode_array_header(io : ByteSource) : ArrayHeader | DBError
diff --git a/pg/pq/connection.iyi b/pg/pq/connection.iyi
index 6fea06e..f34e2d8 100644
--- a/pg/pq/connection.iyi
+++ b/pg/pq/connection.iyi
@@ -21,6 +21,7 @@ import pg/pq/conninfo::{ConnInfo}
 import pg/pq/frame::*
 import pg/pq/param::{Param}
 import pg/pq/statement_cache::{StatementCache}
+import pg/decoder::{Decoders, DecoderTable}
 import pg/pq/scram::{SaslContext, md5_password}
 import pg/pq/ssl::{tls_config, server_end_point}
 import pg/pq/pgpass::{LOG}
@@ -50,6 +51,8 @@ pub class PQConnection
   # The transaction status of the last ReadyForQuery: `'I'` idle, `'T'` in
   # a transaction block, `'E'` in a failed one.
   getter transaction_status : Char = 'I'
+  # The decoders this session reads columns with; see `decoders`.
+  @decoders : DecoderTable? = nil
 
   def initialize(@stream : Stream, @conninfo : ConnInfo)
     @server_parameters = {} of String => String
@@ -70,6 +73,17 @@ pub class PQConnection
   # for the timeout's: a socket's timer fires on the platform's tick.
   TIMEOUT_SLACK_MS = 50
 
+  # The decoders this session reads columns with: the registry's table,
+  # taken again only when a decoder was registered after it, so reading a
+  # column takes no lock.
+  def decoders : DecoderTable
+    table = @decoders
+    return table if table && table.current?(Decoders.generation)
+    table = Decoders.table
+    @decoders = table
+    table
+  end
+
   # The process id of the session's backend on the server, 0 when the
   # server named none.
   def backend_pid : Int32
diff --git a/pg/range.iyi b/pg/range.iyi
index 8053e70..f88c7fc 100644
--- a/pg/range.iyi
+++ b/pg/range.iyi
@@ -144,10 +144,15 @@ pub class RangeDecoder < Decoder
   NO_LOWER = EMPTY | LB_INF | LB_NULL
   NO_UPPER = EMPTY | UB_INF | UB_NULL
 
-  getter oids : Array(Int32)
+  @oids : List(Int32)
   getter subtype_oid : Int32
 
-  def initialize(@oids : Array(Int32), @subtype_oid : Int32)
+  def initialize(oids : Array(Int32), @subtype_oid : Int32)
+    @oids = List(Int32).new(oids)
+  end
+
+  def oids : Array(Int32)
+    @oids.to_a
   end
 
   def type_name : String
@@ -171,13 +176,18 @@ end
 # Reads the binary multirange format: the range count, then each range as
 # its size and the binary range format.
 pub class MultirangeDecoder < Decoder
-  getter oids : Array(Int32)
+  @oids : List(Int32)
   getter range : RangeDecoder
 
-  def initialize(@oids : Array(Int32), subtype_oid : Int32)
+  def initialize(oids : Array(Int32), subtype_oid : Int32)
+    @oids = List(Int32).new(oids)
     @range = RangeDecoder.new([] of Int32, subtype_oid)
   end
 
+  def oids : Array(Int32)
+    @oids.to_a
+  end
+
   def type_name : String
     "PG::MultirangeValue"
   end
diff --git a/pg/record.iyi b/pg/record.iyi
index 36cdcc1..4ef7eec 100644
--- a/pg/record.iyi
+++ b/pg/record.iyi
@@ -9,6 +9,7 @@
 module pg/record
 
 import std/int
+import std/list::{List}
 import db/db/errors::{DBError}
 import db/db/value::{DBValue, CustomValue}
 import pg/pq/stream::{ByteSource}
@@ -47,9 +48,14 @@ end
 pub class RecordDecoder < Decoder
   RECORD_OID = 2249
 
-  getter oids : Array(Int32)
+  @oids : List(Int32)
 
-  def initialize(@oids : Array(Int32) = [RECORD_OID])
+  def initialize(oids : Array(Int32) = [RECORD_OID])
+    @oids = List(Int32).new(oids)
+  end
+
+  def oids : Array(Int32)
+    @oids.to_a
   end
 
   def type_name : String

From e9f9d38d539a6ebebac35c7ac2d509da51ecede1 Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Thu, 29 Oct 2026 00:12:09 +0300
Subject: [PATCH 5/7] Document the statement cache and the Share module state

README: Prepared statements, statement_cache_capacity, Share decoders,
and the crystal-pg differences. CHANGELOG: the unreleased changes, with
the measurements on macOS, and why @[SafeReach] is not used.
---
 CHANGELOG.md | 65 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 README.md    | 61 ++++++++++++++++++++++++++++++++++++++++++++----
 2 files changed, 121 insertions(+), 5 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bcf6ec7..b223e37 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,68 @@
+## Unreleased
+
+A statement cache, and the driver's module state made safe to reach from
+every worker.
+
+* Each connection keeps the statements it runs prepared on the server
+  ([#1](https://github.com/iyilang/pg/issues/1)). A query's first execution
+  on a connection parses it under a name and describes it, in the same
+  round trip as the execution; every later one sends only Bind, Execute and
+  Sync, where each sent Parse, Bind, Describe, Execute and Sync before. The
+  cache is an LRU of `statement_cache_capacity` statements (512, pgx's
+  default), a Close for each one given up going with the next exchange.
+  `statement_cache_capacity=0`, or `prepared_statements_cache=false`, turns
+  it off; unprepared statements are parsed every time, as before.
+  - A kept statement the server refuses to bind as out of date - 26000
+    (dropped), 0A000 (`cached plan must not change result type`), class 42,
+    or class 22 for one with parameters - is prepared anew and the query run
+    again, once, outside a transaction block; inside one the error is
+    answered. `DISCARD ALL` and `DEALLOCATE ALL` empty the cache, and DDL
+    on the same connection (`CREATE`, `ALTER`, `DROP`) gives up every kept
+    statement: one keeps the parameter types inferred when it was prepared,
+    and the suite's `timestamptz` read back three hours off on a server in
+    Europe/Istanbul when its INSERT kept the `timestamp` parameter of the
+    table dropped before.
+  - `PQConnection#statement_cache`, `#parse_count` (the Parse messages a
+    session sent) and `#transaction_status` (the last ReadyForQuery's).
+  - Measured on macOS 26.6, Apple M2 Pro, 10 cores (the user's machine),
+    against a throwaway PostgreSQL 14.19 there, with clients built by iyi
+    master befaa8bc4. A SELECT of four columns by primary key, 1000 times
+    on one connection, sent per query: before, a Parse, a Bind, a Describe
+    portal, an Execute and a Sync; after, a Bind, an Execute and a Sync, and
+    one Parse and one Describe statement in all. The same SELECT in a loop,
+    2 s warm-up then 10 s measured, three runs each, alternating:
+
+    | | queries/s (best of 3) | runs | postgres CPU per query |
+    |---|---:|---|---:|
+    | 1 connection, before | 15,550 | 11,734 / 14,393 / 15,550 | 23.8-27.2 us |
+    | 1 connection, after | 21,739 | 21,739 / 18,728 / 18,945 | 13.7-15.0 us |
+    | 10 connections, before | 52,610 | 52,130 / 52,610 / 51,335 | 35.7-37.0 us |
+    | 10 connections, after | 72,543 | 69,769 / 72,242 / 72,543 | 21.8-22.9 us |
+
+    Client and server shared the machine, which was at 100 % in every run;
+    the client used 659-862 % of it, its collector's helpers spinning as
+    they do on darwin at that iyi master.
+* The module state a connection reaches is `Share`, so a program that opens
+  and queries connections in parallel tasks gets no #148 reach warning from
+  this driver (iyilang/iyi#148). `@[SafeReach]` is not used: it is in no
+  released iyi (not 0.18.0 to 0.19.1), and this driver still builds with
+  iyi 0.18.0. `iyi check` (iyi master befaa8bc4) of a program that opens
+  connections and queries them in `spawn_parallel` tasks warned 12 times,
+  9 of them about this driver's constants; it warns 3 times now, all about
+  iyi-db's (`DRIVERS`, `AROUND_HOOKS`, `LOG`).
+  - `ConnInfo`'s tables (`SSL_MODES`, `KNOWN_KEYS`, `ENV_DEFAULTS`, ...)
+    and `RangeValue::QUOTED` are `List`s.
+  - `LOG` is a `SharedLog`: `Log.for("pg")` behind a `Mutex`, std's `Logger`
+    not being `Share`. Set its level or backend on `Log.for("pg")`.
+  - The decoder registry is a `DecoderRegistry` behind a `Mutex`, and a
+    connection reads its columns through a `DecoderTable` it holds - the
+    OIDs sorted in a `List`, their decoders in another - taken again only
+    after a registration, so reading a column takes no lock. A decoder is
+    `Share`: `ArrayDecoder`, `RecordDecoder`, `RangeDecoder` and
+    `MultirangeDecoder` keep their OIDs in a `List`, and `oids` answers a
+    copy. A decoder of your own sets its fields in `initialize` only, of
+    `Share` types.
+
 ## v0.3.1
 
 On iyi-db 0.3.0, for its pool and closing fixes: `Database#close` closes
diff --git a/README.md b/README.md
index 6268f4d..74fb9cb 100644
--- a/README.md
+++ b/README.md
@@ -67,9 +67,10 @@ end
 The URI's query takes iyi-db's pool options (`max_pool_size`, ...) and the
 connection's: `application_name`, `sslmode`, `sslrootcert`, `sslcert`,
 `sslkey`, `ssl_min_protocol_version`, `channel_binding`, `auth_methods`,
-`connect_timeout`, `options`, `replication`, and `host` for a Unix socket
-directory. A `key=value` string takes the same keys, a value with spaces
-single-quoted (`options='-c search_path=app'`). Missing parts come from
+`connect_timeout`, `options`, `replication`, `statement_cache_capacity`
+(see Prepared statements), and `host` for a Unix socket directory. A
+`key=value` string takes the same keys, a value with spaces single-quoted
+(`options='-c search_path=app'`). Missing parts come from
 `PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER`, `PGPASSWORD`, `PGAPPNAME`,
 `PGSSLMODE`, `PGSSLROOTCERT`, `PGSSLCERT`, `PGSSLKEY`, `PGCHANNELBINDING`,
 `PGSSLMINPROTOCOLVERSION`, `PGCONNECT_TIMEOUT`, `PGOPTIONS` and the password
@@ -87,6 +88,46 @@ command-line options: `options=-c%20search_path%3Dapp`.
 A server error is a `PQError` (module `pg/pq/errors`) whose `fields` hold
 what the server sent: `err.field_message(:code)`, `:detail`, `:hint`, ...
 
+### Prepared statements
+
+A connection keeps the statements it runs prepared on the server, by their
+text. The first execution of a query on a connection parses it under a name
+(`iyi_pg_1`, ...) and describes its rows, in the same round trip as the
+execution; every later one sends only Bind, Execute and Sync, so the server
+neither parses nor plans it again. pgx keeps such a cache; crystal-pg, and
+this driver up to 0.3.1, parse an unnamed statement every time
+([#1](https://github.com/iyilang/pg/issues/1)).
+
+The cache holds `statement_cache_capacity` statements per connection (512,
+pgx's default, unless given), the least recently used one closed on the
+server to make room. A kept statement the server can no longer bind as it
+was prepared - dropped by a `DEALLOCATE`, or out of date after an `ALTER
+TABLE` changed the type of its rows (`cached plan must not change result
+type`) or of a column under one of its parameters - is prepared anew and
+the query run again, once, when that is safe: outside a transaction block.
+Inside one, the transaction is aborted and the error is answered; the
+statement is prepared anew after it. `DISCARD ALL` and `DEALLOCATE ALL`
+empty the cache.
+
+A kept statement keeps the parameter types the server inferred when it was
+prepared. DDL on the same connection (`CREATE ...`, `ALTER ...`,
+`DROP ...`) empties its cache, so its statements are prepared again against
+the new schema. DDL from another session is seen only when a kept statement
+then fails to bind, as above: a column changed to a type an argument
+converts to without an error - `timestamp` to `timestamptz` on a server not
+in UTC, say - leaves a statement prepared before converting to the old
+type, as pgx's do. Run such a migration with the cache off, or reconnect
+after it.
+
+Turn it off with `statement_cache_capacity=0`, or iyi-db's
+`prepared_statements_cache=false` - behind a pooler that does not keep a
+session for a client, say (PgBouncer in transaction mode) - and every
+statement is parsed each time it runs, as the unnamed statement. An
+unprepared statement (`db.unprepared.exec(...)`, or
+`prepared_statements=false`) always is. A connection's cache is
+`pg_conn.connection.statement_cache`, and `pg_conn.connection.parse_count`
+counts the Parse messages it has sent.
+
 ### TLS
 
 Over TCP, a session is encrypted with iyi's `std/tls` as `sslmode` says,
@@ -289,6 +330,13 @@ in its canonical form, `[1,5]` as `[1,6)`. A `MultirangeValue` holds its
 `MultirangeDecoder.new([multirange_oid], subtype_oid)` (module `pg/range`;
 the OIDs are in `pg_range`).
 
+Every thread reads the same decoders, so a `Decoder` of your own is
+`Share`: its fields are set in `initialize` only, and are `Share`
+themselves (a `List`, not an `Array`). A decoder may be registered at any
+time; a connection reads its columns through a table of the registry it
+holds, taken again after a registration, so a column is read without a
+lock.
+
 Values bound as arguments are iyi-db's `DBValue`s: the scalars, `Time` (to
 the microsecond, PostgreSQL's precision), `Span`, `Bytes`, arrays, and this
 driver's value classes (`Numeric`, `Interval`, the geometric types,
@@ -344,7 +392,7 @@ mistake that postgres continues to have this as an option at all.
 |---|---|
 | `pg` (the root module) | `Pg.connect`, `connect_listen`, `connect_replication`, `ListenConnection`, `VERSION` |
 | `pg/core` | `PGDriver`, `PGConnection`, `PGStatement`, `PGResultSet`, `PGConnectionBuilder`; registers the driver |
-| `pg/decoder` | `Decoder` and one per type, `Decoders` (`from_oid`, `register_decoder`), `UUIDValue`, `JSONValue` |
+| `pg/decoder` | `Decoder` and one per type, `Decoders` (`from_oid`, `register_decoder`), `DecoderTable`, `UUIDValue`, `JSONValue` |
 | `pg/numeric`, `pg/interval`, `pg/geo`, `pg/record`, `pg/time_tz`, `pg/inet` | the value classes |
 | `pg/range` | `RangeValue`, `MultirangeValue`, `RangeDecoder`, `MultirangeDecoder` |
 | `pg/escape` | `EscapeHelper`: `escape_identifier`, `escape_literal` |
@@ -353,7 +401,8 @@ mistake that postgres continues to have this as an option at all.
 | `pg/replication/messages` | the replication frames and WAL messages: `XLogData`, `Begin`, `Insert`, ... |
 | `pg/errors` | `PGError`, `RuntimeError` |
 | `pg/pq/connection` | `PQConnection`, the protocol session |
-| `pg/pq/conninfo`, `pg/pq/pgpass` | `ConnInfo`, `PgPass` |
+| `pg/pq/conninfo`, `pg/pq/pgpass` | `ConnInfo`, `PgPass`; `LOG`, the driver's `SharedLog` |
+| `pg/pq/statement_cache` | `StatementCache`, a session's prepared statements, and `CachedStatement` |
 | `pg/pq/ssl` | the `TlsConfig` a `ConnInfo` asks for: roots, client certificate; the channel binding's certificate hash |
 | `pg/pq/frame`, `pg/pq/param`, `pg/pq/stream`, `pg/pq/scram` | messages, parameter encoding, the buffered socket or TLS session, SCRAM |
 | `pg/pq/errors` | `ConnectionError`, `StreamError`, `PQError`, `ErrorField` |
@@ -374,6 +423,8 @@ mistake that postgres continues to have this as an option at all.
 | `CopyResult < IO` | `CopyResult` with `read`, `read_char`, `gets_to_end`, `write`, `<<`, `close` answering errors | `IyiIO` cannot answer one |
 | enums bound as their value | pass `.value` | an enum is not a `DBValue` |
 | `sslmode=require` and `prefer` encrypt without checking the certificate; `verify-ca` checks no host name | every certificate is checked, chain and host name: against `sslrootcert` or `root.crt`, else the system's roots | `std/tls` has no way to turn the checks off |
+| every execution parses an unnamed statement: Parse, Bind, Describe, Execute, Sync | a statement is kept prepared per connection after its first execution: Bind, Execute, Sync | the server parsed and planned each query again ([#1](https://github.com/iyilang/pg/issues/1)) |
+| `PQ::Log`, a `::Log` | `LOG` is a `SharedLog`, a `Logger` behind a `Mutex`; configure `Log.for("pg")` | std's `Logger` is not `Share`, and a connection logs from any worker |
 
 ## Development
 

From d296282b14827b9b21b8dc1e184f60324b7539d2 Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Mon, 2 Nov 2026 11:52:23 +0300
Subject: [PATCH 6/7] Move to iyi-db 0.4.0 and iyi 0.20.0

---
 .github/workflows/ci.yml | 2 +-
 CHANGELOG.md             | 9 ++++++---
 README.md                | 2 +-
 iyi.mod                  | 2 +-
 iyi.sum                  | 2 +-
 5 files changed, 10 insertions(+), 7 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0741017..9fd4fd0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -23,7 +23,7 @@ jobs:
           --health-retries 10
     env:
       # The oldest iyi this release supports (README, Requirements).
-      IYI_VERSION: 0.18.0
+      IYI_VERSION: 0.20.0
       DATABASE_URL: postgres://postgres:postgres@127.0.0.1/iyi_pg_test
     steps:
       - uses: actions/checkout@v4
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b223e37..f5dd94a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,10 @@
 A statement cache, and the driver's module state made safe to reach from
 every worker.
 
+Requires iyi 0.20.0 and iyi-db 0.4.0, whose `Database` and pool are `Share`
+and may be used by parallel tasks on every worker; iyi-db 0.4.0 requires
+iyi 0.20.0.
+
 * Each connection keeps the statements it runs prepared on the server
   ([#1](https://github.com/iyilang/pg/issues/1)). A query's first execution
   on a connection parses it under a name and describes it, in the same
@@ -44,9 +48,8 @@ every worker.
     they do on darwin at that iyi master.
 * The module state a connection reaches is `Share`, so a program that opens
   and queries connections in parallel tasks gets no #148 reach warning from
-  this driver (iyilang/iyi#148). `@[SafeReach]` is not used: it is in no
-  released iyi (not 0.18.0 to 0.19.1), and this driver still builds with
-  iyi 0.18.0. `iyi check` (iyi master befaa8bc4) of a program that opens
+  this driver (iyilang/iyi#148). `@[SafeReach]`, new in iyi 0.20.0, is
+  not used. `iyi check` (iyi master befaa8bc4) of a program that opens
   connections and queries them in `spawn_parallel` tasks warned 12 times,
   9 of them about this driver's constants; it warns 3 times now, all about
   iyi-db's (`DRIVERS`, `AROUND_HOOKS`, `LOG`).
diff --git a/README.md b/README.md
index 74fb9cb..dfe8444 100644
--- a/README.md
+++ b/README.md
@@ -14,7 +14,7 @@ are closed, and tasks are structured.
 
 ## Requirements
 
-iyi 0.18.0 or later, and iyi-db 0.3.0 (required by this package's `iyi.mod`).
+iyi 0.20.0 or later, and iyi-db 0.4.0 (required by this package's `iyi.mod`).
 Tested against PostgreSQL 16.
 
 ## Installation
diff --git a/iyi.mod b/iyi.mod
index 95ad61a..42abb31 100644
--- a/iyi.mod
+++ b/iyi.mod
@@ -1,4 +1,4 @@
 # `iyi get example.com/someone/lib` adds a dependency here.
 module github.com/iyilang/pg
 
-require github.com/iyilang/db v0.3.0 as db
+require github.com/iyilang/db v0.4.0 as db
diff --git a/iyi.sum b/iyi.sum
index c4d3cee..e1de698 100644
--- a/iyi.sum
+++ b/iyi.sum
@@ -1 +1 @@
-github.com/iyilang/db v0.3.0 s1:1a1325d1be2abc9d8bafd193ba35192f9c041a39
+github.com/iyilang/db v0.4.0 s1:eb713fe8bccc5b89be8d2efa16e515ae96667a15

From 670407ed60e00cbf79fc6cf237fbd2ac0becbbba Mon Sep 17 00:00:00 2001
From: Serdar Dogruyol <990485+sdogruyol@users.noreply.github.com>
Date: Mon, 2 Nov 2026 11:52:23 +0300
Subject: [PATCH 7/7] Release v0.4.0

---
 CHANGELOG.md | 2 +-
 pg.iyi       | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f5dd94a..d4d6617 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,4 @@
-## Unreleased
+## v0.4.0
 
 A statement cache, and the driver's module state made safe to reach from
 every worker.
diff --git a/pg.iyi b/pg.iyi
index 02cd634..b40446c 100644
--- a/pg.iyi
+++ b/pg.iyi
@@ -24,7 +24,7 @@ import pg/core::{PGConnection}
 import std/time::{Span}
 import pg/replication::{Handler}
 
-pub VERSION = "0.3.1"
+pub VERSION = "0.4.0"
 
 # Establish a connection pool to the database.
 pub def connect(url : String) : Database | DBError