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 bcf6ec7..d4d6617 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 6268f4d..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 @@ -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 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 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 diff --git a/pg/core.iyi b/pg/core.iyi index f85291e..4fb99fa 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} @@ -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,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/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 ae4d512..f34e2d8 100644 --- a/pg/pq/connection.iyi +++ b/pg/pq/connection.iyi @@ -20,6 +20,8 @@ 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/decoder::{Decoders, DecoderTable} import pg/pq/scram::{SaslContext, md5_password} import pg/pq/ssl::{tls_config, server_end_point} import pg/pq/pgpass::{LOG} @@ -41,11 +43,22 @@ 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' + # The decoders this session reads columns with; see `decoders`. + @decoders : DecoderTable? = nil 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 @@ -60,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 @@ -341,6 +365,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 +397,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 +740,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 +785,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..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,14 @@ 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. + STATEMENT_CACHE_CAPACITY = 512 # The host. If it starts with a / it is the directory of a local Unix # socket. @@ -101,6 +108,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 +138,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 +225,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 +247,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 +341,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/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/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/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/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/range.iyi b/pg/range.iyi index 7de722f..f88c7fc 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 @@ -142,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 @@ -169,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 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 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