A per-connection prepared-statement cache, and constants every worker may read

.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

CHANGELOG.md

@@ -1,3 +1,71 @@
+## v0.4.0
+
+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
+  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]`, 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`).
+  - `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

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
@@ -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
 

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

iyi.sum

@@ -1 +1 @@
-github.com/iyilang/db v0.3.0 s1:1a1325d1be2abc9d8bafd193ba35192f9c041a39
+github.com/iyilang/db v0.4.0 s1:eb713fe8bccc5b89be8d2efa16e515ae96667a15

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

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}
@@ -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
@@ -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,31 +392,121 @@ 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
     @conn.send_execute_message
     @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