Revert "Require a reason when marking a server bad (#654)"

This reverts commit 4dbef49ec9.

.circleci/config.yml

@@ -9,7 +9,7 @@ jobs:
     # Specify the execution environment. You can specify an image from Dockerhub or use one of our Convenience Images from CircleCI's Developer Hub.
     # See: https://circleci.com/docs/2.0/configuration-reference/#docker-machine-macos-windows-executor
     docker:
-      - image: ghcr.io/levkk/pgcat-ci:1.67
+      - image: ghcr.io/postgresml/pgcat-ci:latest
         environment:
           RUST_LOG: info
           LLVM_PROFILE_FILE: /tmp/pgcat-%m-%p.profraw
@@ -63,6 +63,9 @@ jobs:
       - run:
           name: "Lint"
           command: "cargo fmt --check"
+      - run:
+          name: "Clippy"
+          command: "cargo clippy --all --all-targets -- -Dwarnings"
       - run:
           name: "Tests"
           command: "cargo clean && cargo build && cargo test && bash .circleci/run_tests.sh && .circleci/generate_coverage.sh"

.circleci/pgcat.toml

@@ -74,6 +74,10 @@ default_role = "any"
 # we'll direct it to the primary.
 query_parser_enabled = true
 
+# If the query parser is enabled and this setting is enabled, we'll attempt to
+# infer the role from the query itself.
+query_parser_read_write_splitting = true
+
 # If the query parser is enabled and this setting is enabled, the primary will be part of the pool of databases used for
 # load balancing of read queries. Otherwise, the primary will only be used for write
 # queries. The primary can always be explicitely selected with our custom protocol.
@@ -134,6 +138,7 @@ database = "shard2"
 pool_mode = "session"
 default_role = "primary"
 query_parser_enabled = true
+query_parser_read_write_splitting = true
 primary_reads_enabled = true
 sharding_function = "pg_bigint_hash"
 

.circleci/run_tests.sh

@@ -108,8 +108,24 @@ cd ../..
 pip3 install -r tests/python/requirements.txt
 python3 tests/python/tests.py || exit 1
 
+
+#
+# Go tests
+# Starts its own pgcat server
+#
+pushd tests/go
+/usr/local/go/bin/go test || exit 1
+popd
+
 start_pgcat "info"
 
+#
+# Rust tests
+#
+cd tests/rust
+cargo run
+cd ../../
+
 # Admin tests
 export PGPASSWORD=admin_pass
 psql -U admin_user -e -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW STATS' > /dev/null

.github/workflows/build-and-push.yaml

@@ -1,6 +1,11 @@
 name: Build and Push
 
-on: push
+on:
+  push:
+    branches:    
+      - main
+    tags:
+      - v*
 
 env:
   registry: ghcr.io
@@ -29,6 +34,7 @@ jobs:
           tags: |
             type=sha,prefix=,format=long
             type=schedule
+            type=ref,event=tag
             type=ref,event=branch
             type=ref,event=pr
             type=raw,value=latest,enable={{ is_default_branch }}

.github/workflows/publish-deb-package.yml

@@ -0,0 +1,48 @@
+name: pgcat package (deb)
+
+on:
+  workflow_dispatch:
+    inputs:
+      packageVersion:
+        default: "1.1.2-dev1"
+jobs:
+  build:
+    strategy:
+      max-parallel: 1
+      fail-fast: false # Let the other job finish, or they can lock each other out
+      matrix:
+        os: ["buildjet-4vcpu-ubuntu-2204", "buildjet-4vcpu-ubuntu-2204-arm"]
+
+    runs-on: ${{ matrix.os }}
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions-rs/toolchain@v1
+      with:
+        toolchain: stable
+    - name: Install dependencies
+      env:
+        DEBIAN_FRONTEND: noninteractive
+        TZ: Etc/UTC
+      run: |
+        curl -sLO https://github.com/deb-s3/deb-s3/releases/download/0.11.4/deb-s3-0.11.4.gem
+        sudo gem install deb-s3-0.11.4.gem
+        dpkg-deb --version
+    - name: Build and release package
+      env:
+        AWS_ACCESS_KEY_ID: ${{ vars.AWS_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+        AWS_DEFAULT_REGION: ${{ vars.AWS_DEFAULT_REGION }}
+      run: |
+        if [[ $(arch) == "x86_64" ]]; then
+          export ARCH=amd64
+        else
+          export ARCH=arm64
+        fi
+
+        bash utilities/deb.sh ${{ inputs.packageVersion }}
+
+        deb-s3 upload \
+          --lock \
+          --bucket apt.postgresml.org \
+          pgcat-${{ inputs.packageVersion }}-ubuntu22.04-${ARCH}.deb \
+          --codename $(lsb_release -cs)

.gitignore

@@ -10,3 +10,4 @@ lcov.info
 dev/.bash_history
 dev/cache
 !dev/cache/.keepme
+.venv

CONFIG.md

@@ -1,4 +1,4 @@
-# PgCat Configurations 
+# PgCat Configurations
 ## `general` Section
 
 ### host
@@ -57,6 +57,38 @@ default: 86400000 # 24 hours
 
 Max connection lifetime before it's closed, even if actively used.
 
+### server_round_robin
+```
+path: general.server_round_robin
+default: false
+```
+
+Whether to use round robin for server selection or not.
+
+### server_tls
+```
+path: general.server_tls
+default: false
+```
+
+Whether to use TLS for server connections or not.
+
+### verify_server_certificate
+```
+path: general.verify_server_certificate
+default: false
+```
+
+Whether to verify server certificate or not.
+
+### verify_config
+```
+path: general.verify_config
+default: true
+```
+
+Whether to verify config or not.
+
 ### idle_client_in_transaction_timeout
 ```
 path: general.idle_client_in_transaction_timeout
@@ -116,10 +148,10 @@ If we should log client disconnections
 ### autoreload
 ```
 path: general.autoreload
-default: 15000
+default: 15000 # milliseconds
 ```
 
-When set to true, PgCat reloads configs if it detects a change in the config file.
+When set, PgCat automatically reloads its configurations at the specified interval (in milliseconds) if it detects changes in the configuration file. The default interval is 15000 milliseconds or 15 seconds.
 
 ### worker_threads
 ```
@@ -151,7 +183,13 @@ path: general.tcp_keepalives_interval
 default: 5
 ```
 
-Number of seconds between keepalive packets.
+### tcp_user_timeout
+```
+path: general.tcp_user_timeout
+default: 10000
+```
+A linux-only parameters that defines the amount of time in milliseconds that transmitted data may remain unacknowledged or buffered data may remain untransmitted (due to zero window size) before TCP will forcibly disconnect
+
 
 ### tls_certificate
 ```
@@ -188,6 +226,39 @@ default: "admin_pass"
 
 Password to access the virtual administrative database
 
+### auth_query
+```
+path: general.auth_query
+default: <UNSET>
+example: "SELECT $1"
+```
+
+Query to be sent to servers to obtain the hash used for md5 authentication. The connection will be
+established using the database configured in the pool. This parameter is inherited by every pool
+and can be redefined in pool configuration.
+
+### auth_query_user
+```
+path: general.auth_query_user
+default: <UNSET>
+example: "sharding_user"
+```
+
+User to be used for connecting to servers to obtain the hash used for md5 authentication by sending the query
+specified in `auth_query_user`. The connection will be established using the database configured in the pool.
+This parameter is inherited by every pool and can be redefined in pool configuration.
+
+### auth_query_password
+```
+path: general.auth_query_password
+default: <UNSET>
+example: "sharding_user"
+```
+
+Password to be used for connecting to servers to obtain the hash used for md5 authentication by sending the query
+specified in `auth_query_user`. The connection will be established using the database configured in the pool.
+This parameter is inherited by every pool and can be redefined in pool configuration.
+
 ### dns_cache_enabled
 ```
 path: general.dns_cache_enabled
@@ -224,7 +295,7 @@ default: "random"
 
 Load balancing mode
 `random` selects the server at random
-`loc` selects the server with the least outstanding busy conncetions
+`loc` selects the server with the least outstanding busy connections
 
 ### default_role
 ```
@@ -237,6 +308,15 @@ If the client doesn't specify, PgCat routes traffic to this role by default.
 `replica` round-robin between replicas only without touching the primary,
 `primary` all queries go to the primary unless otherwise specified.
 
+### prepared_statements_cache_size
+```
+path: general.prepared_statements_cache_size
+default: 0
+```
+
+Size of the prepared statements cache. 0 means disabled.
+TODO: update documentation
+
 ### query_parser_enabled
 ```
 path: pools.<pool_name>.query_parser_enabled

CONTRIBUTING.md

@@ -2,7 +2,7 @@
 
 Thank you for contributing! Just a few tips here:
 
-1. `cargo fmt` your code before opening up a PR
+1. `cargo fmt` and `cargo clippy` your code before opening up a PR
 2. Run the test suite (e.g. `pgbench`) to make sure everything still works. The tests are in `.circleci/run_tests.sh`.
 3. Performance is important, make sure there are no regressions in your branch vs. `main`.
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgcat"
-version = "1.0.2-alpha3"
+version = "1.1.2-dev4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -8,7 +8,7 @@ edition = "2021"
 tokio = { version = "1", features = ["full"] }
 bytes = "1"
 md-5 = "0.10"
-bb8 = "0.8.0"
+bb8 = "0.8.1"
 async-trait = "0.1"
 rand = "0.8"
 chrono = "0.4"
@@ -19,10 +19,9 @@ serde_derive = "1"
 regex = "1"
 num_cpus = "1"
 once_cell = "1"
-sqlparser = {version = "0.33", features = ["visitor"] }
+sqlparser = {version = "0.34", features = ["visitor"] }
 log = "0.4"
 arc-swap = "1"
-env_logger = "0.10"
 parking_lot = "0.12.1"
 hmac = "0.12"
 sha2 = "0.10"
@@ -34,7 +33,7 @@ hyper = { version = "0.14", features = ["full"] }
 phf = { version = "0.11.1", features = ["macros"] }
 exitcode = "1.1.2"
 futures = "0.3"
-socket2 = { version = "0.5.3", features = ["all"] }
+socket2 = { version = "0.4.7", features = ["all"] }
 nix = "0.26.2"
 atomic_enum = "0.2.0"
 postgres-protocol = "0.6.5"
@@ -46,6 +45,10 @@ trust-dns-resolver = "0.22.0"
 tokio-test = "0.4.2"
 serde_json = "1"
 itertools = "0.10"
+clap = { version = "4.3.1", features = ["derive", "env"] }
+tracing = "0.1.37"
+tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter", "std"]}
+lru = "0.12.0"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 jemallocator = "0.5.0"

Dockerfile

@@ -1,9 +1,19 @@
-FROM rust:1 AS builder
+FROM rust:1-slim-bookworm AS builder
+
+RUN apt-get update && \
+    apt-get install -y build-essential
+
 COPY . /app
 WORKDIR /app
 RUN cargo build --release
 
-FROM debian:bullseye-slim
+FROM debian:bookworm-slim
+RUN apt-get update && apt-get install  -o Dpkg::Options::=--force-confdef -yq --no-install-recommends \
+    postgresql-client \
+    # Clean up layer
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* \
+    && truncate -s 0 /var/log/*log
 COPY --from=builder /app/target/release/pgcat /usr/bin/pgcat
 COPY --from=builder /app/pgcat.toml /etc/pgcat/pgcat.toml
 WORKDIR /etc/pgcat

Dockerfile.ci

@@ -1,4 +1,6 @@
 FROM cimg/rust:1.67.1
+COPY --from=sclevine/yj /bin/yj /bin/yj
+RUN /bin/yj -h
 RUN sudo apt-get update && \
 	sudo apt-get install -y \
 		psmisc postgresql-contrib-14 postgresql-client-14 libpq-dev \
@@ -7,6 +9,9 @@ RUN sudo apt-get update && \
 	sudo apt-get upgrade curl && \
 	cargo install cargo-binutils rustfilt && \
 	rustup component add llvm-tools-preview && \
- 	pip3 install psycopg2 && sudo gem install bundler && \
+	pip3 install psycopg2 && sudo gem install bundler && \
 	wget -O /tmp/toxiproxy-2.4.0.deb https://github.com/Shopify/toxiproxy/releases/download/v2.4.0/toxiproxy_2.4.0_linux_$(dpkg --print-architecture).deb && \
 	sudo dpkg -i /tmp/toxiproxy-2.4.0.deb
+RUN wget -O /tmp/go1.21.3.linux-$(dpkg --print-architecture).tar.gz https://go.dev/dl/go1.21.3.linux-$(dpkg --print-architecture).tar.gz && \
+    sudo tar -C /usr/local -xzf /tmp/go1.21.3.linux-$(dpkg --print-architecture).tar.gz && \
+    rm /tmp/go1.21.3.linux-$(dpkg --print-architecture).tar.gz

Dockerfile.dev

@@ -0,0 +1,25 @@
+FROM lukemathwalker/cargo-chef:latest-rust-1 AS chef
+
+RUN apt-get update && \
+    apt-get install -y build-essential
+
+WORKDIR /app
+
+FROM chef AS planner
+COPY . .
+RUN cargo chef prepare --recipe-path recipe.json
+
+FROM chef AS builder 
+COPY --from=planner /app/recipe.json recipe.json
+# Build dependencies - this is the caching Docker layer!
+RUN cargo chef cook --release --recipe-path recipe.json
+# Build application
+COPY . .
+RUN cargo build
+
+FROM debian:bookworm-slim
+COPY --from=builder /app/target/release/pgcat /usr/bin/pgcat
+COPY --from=builder /app/pgcat.toml /etc/pgcat/pgcat.toml
+WORKDIR /etc/pgcat
+ENV RUST_LOG=info
+CMD ["pgcat"]

README.md

@@ -40,7 +40,7 @@ PgCat is stable and used in production to serve hundreds of thousands of queries
       </a>
     </td>
     <td>
-      <a href="https://postgresml.org/blog/scaling-postgresml-to-one-million-requests-per-second">
+      <a href="https://postgresml.org/blog/scaling-postgresml-to-1-million-requests-per-second">
         <img src="./images/postgresml.webp" height="70" width="auto">
       </a>
     </td>
@@ -57,7 +57,7 @@ PgCat is stable and used in production to serve hundreds of thousands of queries
       </a>
     </td>
     <td>
-      <a href="https://postgresml.org/blog/scaling-postgresml-to-one-million-requests-per-second">
+      <a href="https://postgresml.org/blog/scaling-postgresml-to-1-million-requests-per-second">
         PostgresML
       </a>
     </td>

control

@@ -0,0 +1,9 @@
+Package: pgcat
+Version: ${PACKAGE_VERSION}
+Section: database
+Priority: optional
+Architecture: ${ARCH}
+Maintainer: PostgresML <team@postgresml.org>
+Homepage: https://postgresml.org
+Description: PgCat - NextGen PostgreSQL Pooler
+ PostgreSQL pooler and proxy (like PgBouncer) with support for sharding, load balancing, failover and mirroring.

dev/Dockerfile

@@ -1,6 +1,8 @@
-FROM rust:bullseye
+FROM rust:1.70-bullseye
 
 # Dependencies
+COPY --from=sclevine/yj /bin/yj /bin/yj
+RUN /bin/yj -h
 RUN apt-get update -y \
     && apt-get install -y \
     llvm-11 psmisc postgresql-contrib postgresql-client \

examples/docker/pgcat.toml

@@ -71,6 +71,10 @@ default_role = "any"
 # we'll direct it to the primary.
 query_parser_enabled = true
 
+# If the query parser is enabled and this setting is enabled, we'll attempt to
+# infer the role from the query itself.
+query_parser_read_write_splitting = true
+
 # If the query parser is enabled and this setting is enabled, the primary will be part of the pool of databases used for
 # load balancing of read queries. Otherwise, the primary will only be used for write
 # queries. The primary can always be explicitly selected with our custom protocol.

pgcat.service

@@ -0,0 +1,16 @@
+[Unit]
+Description=PgCat pooler
+After=network.target
+StartLimitIntervalSec=0
+
+[Service]
+User=pgcat
+Type=simple
+Restart=always
+RestartSec=1
+Environment=RUST_LOG=info
+LimitNOFILE=65536
+ExecStart=/usr/bin/pgcat /etc/pgcat.toml
+
+[Install]
+WantedBy=multi-user.target

pgcat.toml

@@ -150,12 +150,20 @@ load_balancing_mode = "random"
 # `primary` all queries go to the primary unless otherwise specified.
 default_role = "any"
 
+# Prepared statements cache size.
+# TODO: update documentation
+prepared_statements_cache_size = 500
+
 # If Query Parser is enabled, we'll attempt to parse
 # every incoming query to determine if it's a read or a write.
 # If it's a read query, we'll direct it to a replica. Otherwise, if it's a write,
 # we'll direct it to the primary.
 query_parser_enabled = true
 
+# If the query parser is enabled and this setting is enabled, we'll attempt to
+# infer the role from the query itself.
+query_parser_read_write_splitting = true
+
 # If the query parser is enabled and this setting is enabled, the primary will be part of the pool of databases used for
 # load balancing of read queries. Otherwise, the primary will only be used for write
 # queries. The primary can always be explicitly selected with our custom protocol.
@@ -167,6 +175,12 @@ primary_reads_enabled = true
 # shard_id_regex = '/\* shard_id: (\d+) \*/'
 # regex_search_limit = 1000 # only look at the first 1000 characters of SQL statements
 
+# Defines the behavior when no shard is selected in a sharded system.
+# `random`: picks a shard at random
+# `random_healthy`: picks a shard at random favoring shards with the least number of recent errors
+# `shard_<number>`: e.g. shard_0, shard_4, etc. picks a specific shard, everytime
+# no_shard_specified_behavior = "shard_0"
+
 # So what if you wanted to implement a different hashing function,
 # or you've already built one and you want this pooler to use it?
 # Current options:
@@ -177,7 +191,7 @@ sharding_function = "pg_bigint_hash"
 # Query to be sent to servers to obtain the hash used for md5 authentication. The connection will be
 # established using the database configured in the pool. This parameter is inherited by every pool
 # and can be redefined in pool configuration.
-# auth_query = "SELECT $1"
+# auth_query="SELECT usename, passwd FROM pg_shadow WHERE usename='$1'"
 
 # User to be used for connecting to servers to obtain the hash used for md5 authentication by sending the query
 # specified in `auth_query_user`. The connection will be established using the database configured in the pool.
@@ -264,7 +278,7 @@ username = "sharding_user"
 # if `server_password` is not set.
 password = "sharding_user"
 
-pool_mode = "session"
+pool_mode = "transaction"
 
 # PostgreSQL username used to connect to the server.
 # server_username = "another_user"
@@ -287,6 +301,8 @@ username = "other_user"
 password = "other_user"
 pool_size = 21
 statement_timeout = 15000
+connect_timeout = 1000
+idle_timeout = 1000
 
 # Shard configs are structured as pool.<pool_name>.shards.<shard_id>
 # Each shard config contains a list of servers that make up the shard

postinst

@@ -0,0 +1,9 @@
+#!/bin/bash
+set -e
+
+systemctl daemon-reload
+systemctl enable pgcat
+
+if ! id pgcat 2> /dev/null; then
+	useradd -s /usr/bin/false pgcat
+fi

postrm

@@ -0,0 +1,4 @@
+#!/bin/bash
+set -e
+
+systemctl daemon-reload

prerm

@@ -0,0 +1,5 @@
+#!/bin/bash
+set -e
+
+systemctl stop pgcat
+systemctl disable pgcat

src/admin.rs

@@ -1,4 +1,6 @@
 use crate::pool::BanReason;
+use crate::server::ServerParameters;
+use crate::stats::pool::PoolStats;
 use bytes::{Buf, BufMut, BytesMut};
 use log::{error, info, trace};
 use nix::sys::signal::{self, Signal};
@@ -14,18 +16,18 @@ use crate::errors::Error;
 use crate::messages::*;
 use crate::pool::ClientServerMap;
 use crate::pool::{get_all_pools, get_pool};
-use crate::stats::{get_client_stats, get_pool_stats, get_server_stats, ClientState, ServerState};
+use crate::stats::{get_client_stats, get_server_stats, ClientState, ServerState};
 
-pub fn generate_server_info_for_admin() -> BytesMut {
-    let mut server_info = BytesMut::new();
+pub fn generate_server_parameters_for_admin() -> ServerParameters {
+    let mut server_parameters = ServerParameters::new();
 
-    server_info.put(server_parameter_message("application_name", ""));
-    server_info.put(server_parameter_message("client_encoding", "UTF8"));
-    server_info.put(server_parameter_message("server_encoding", "UTF8"));
-    server_info.put(server_parameter_message("server_version", VERSION));
-    server_info.put(server_parameter_message("DateStyle", "ISO, MDY"));
+    server_parameters.set_param("application_name".to_string(), "".to_string(), true);
+    server_parameters.set_param("client_encoding".to_string(), "UTF8".to_string(), true);
+    server_parameters.set_param("server_encoding".to_string(), "UTF8".to_string(), true);
+    server_parameters.set_param("server_version".to_string(), VERSION.to_string(), true);
+    server_parameters.set_param("DateStyle".to_string(), "ISO, MDY".to_string(), true);
 
-    server_info
+    server_parameters
 }
 
 /// Handle admin client.
@@ -72,17 +74,21 @@ where
         }
         "PAUSE" => {
             trace!("PAUSE");
-            pause(stream, query_parts[1]).await
+            pause(stream, query_parts).await
         }
         "RESUME" => {
             trace!("RESUME");
-            resume(stream, query_parts[1]).await
+            resume(stream, query_parts).await
         }
         "SHUTDOWN" => {
             trace!("SHUTDOWN");
             shutdown(stream).await
         }
         "SHOW" => match query_parts[1].to_ascii_uppercase().as_str() {
+            "HELP" => {
+                trace!("SHOW HELP");
+                show_help(stream).await
+            }
             "BANS" => {
                 trace!("SHOW BANS");
                 show_bans(stream).await
@@ -254,39 +260,50 @@ async fn show_pools<T>(stream: &mut T) -> Result<(), Error>
 where
     T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let all_pool_stats = get_pool_stats();
+    let pool_lookup = PoolStats::construct_pool_lookup();
+    let mut res = BytesMut::new();
+    res.put(row_description(&PoolStats::generate_header()));
+    pool_lookup.iter().for_each(|(_identifier, pool_stats)| {
+        res.put(data_row(&pool_stats.generate_row()));
+    });
+    res.put(command_complete("SHOW"));
 
-    let columns = vec![
-        ("database", DataType::Text),
-        ("user", DataType::Text),
-        ("pool_mode", DataType::Text),
-        ("cl_idle", DataType::Numeric),
-        ("cl_active", DataType::Numeric),
-        ("cl_waiting", DataType::Numeric),
-        ("cl_cancel_req", DataType::Numeric),
-        ("sv_active", DataType::Numeric),
-        ("sv_idle", DataType::Numeric),
-        ("sv_used", DataType::Numeric),
-        ("sv_tested", DataType::Numeric),
-        ("sv_login", DataType::Numeric),
-        ("maxwait", DataType::Numeric),
-        ("maxwait_us", DataType::Numeric),
+    // ReadyForQuery
+    res.put_u8(b'Z');
+    res.put_i32(5);
+    res.put_u8(b'I');
+
+    write_all_half(stream, &res).await
+}
+
+/// Show all available options.
+async fn show_help<T>(stream: &mut T) -> Result<(), Error>
+where
+    T: tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    let mut res = BytesMut::new();
+
+    let detail_msg = [
+        "",
+        "SHOW HELP|CONFIG|DATABASES|POOLS|CLIENTS|SERVERS|USERS|VERSION",
+        // "SHOW PEERS|PEER_POOLS", // missing PEERS|PEER_POOLS
+        // "SHOW FDS|SOCKETS|ACTIVE_SOCKETS|LISTS|MEM|STATE", // missing FDS|SOCKETS|ACTIVE_SOCKETS|MEM|STATE
+        "SHOW LISTS",
+        // "SHOW DNS_HOSTS|DNS_ZONES", // missing DNS_HOSTS|DNS_ZONES
+        "SHOW STATS", // missing STATS_TOTALS|STATS_AVERAGES|TOTALS
+        "SET key = arg",
+        "RELOAD",
+        "PAUSE [<db>, <user>]",
+        "RESUME [<db>, <user>]",
+        // "DISABLE <db>", // missing
+        // "ENABLE <db>", // missing
+        // "RECONNECT [<db>]", missing
+        // "KILL <db>",
+        // "SUSPEND",
+        "SHUTDOWN",
     ];
 
-    let mut res = BytesMut::new();
-    res.put(row_description(&columns));
-
-    for ((_user_pool, _pool), pool_stats) in all_pool_stats {
-        let mut row = vec![
-            pool_stats.database(),
-            pool_stats.user(),
-            pool_stats.pool_mode().to_string(),
-        ];
-        pool_stats.populate_row(&mut row);
-        pool_stats.clear_maxwait();
-        res.put(data_row(&row));
-    }
-
+    res.put(notify("Console usage", detail_msg.join("\n\t")));
     res.put(command_complete("SHOW"));
 
     // ReadyForQuery
@@ -334,17 +351,17 @@ where
                 let paused = pool.paused();
 
                 res.put(data_row(&vec![
-                    address.name(),                         // name
-                    address.host.to_string(),               // host
-                    address.port.to_string(),               // port
-                    database_name.to_string(),              // database
-                    pool_config.user.username.to_string(),  // force_user
-                    pool_config.user.pool_size.to_string(), // pool_size
-                    "0".to_string(),                        // min_pool_size
-                    "0".to_string(),                        // reserve_pool
-                    pool_config.pool_mode.to_string(),      // pool_mode
-                    pool_config.user.pool_size.to_string(), // max_connections
-                    pool_state.connections.to_string(),     // current_connections
+                    address.name(),                                          // name
+                    address.host.to_string(),                                // host
+                    address.port.to_string(),                                // port
+                    database_name.to_string(),                               // database
+                    pool_config.user.username.to_string(),                   // force_user
+                    pool_config.user.pool_size.to_string(),                  // pool_size
+                    pool_config.user.min_pool_size.unwrap_or(0).to_string(), // min_pool_size
+                    "0".to_string(),                                         // reserve_pool
+                    pool_config.pool_mode.to_string(),                       // pool_mode
+                    pool_config.user.pool_size.to_string(),                  // max_connections
+                    pool_state.connections.to_string(),                      // current_connections
                     match paused {
                         // paused
                         true => "1".to_string(),
@@ -673,6 +690,8 @@ where
         ("query_count", DataType::Numeric),
         ("error_count", DataType::Numeric),
         ("age_seconds", DataType::Numeric),
+        ("maxwait", DataType::Numeric),
+        ("maxwait_us", DataType::Numeric),
     ];
 
     let new_map = get_client_stats();
@@ -680,6 +699,7 @@ where
     res.put(row_description(&columns));
 
     for (_, client) in new_map {
+        let max_wait = client.max_wait_time.load(Ordering::Relaxed);
         let row = vec![
             format!("{:#010X}", client.client_id()),
             client.pool_name(),
@@ -693,6 +713,8 @@ where
                 .duration_since(client.connect_time())
                 .as_secs()
                 .to_string(),
+            (max_wait / 1_000_000).to_string(),
+            (max_wait % 1_000_000).to_string(),
         ];
 
         res.put(data_row(&row));
@@ -725,6 +747,10 @@ where
         ("bytes_sent", DataType::Numeric),
         ("bytes_received", DataType::Numeric),
         ("age_seconds", DataType::Numeric),
+        ("prepare_cache_hit", DataType::Numeric),
+        ("prepare_cache_miss", DataType::Numeric),
+        ("prepare_cache_eviction", DataType::Numeric),
+        ("prepare_cache_size", DataType::Numeric),
     ];
 
     let new_map = get_server_stats();
@@ -748,6 +774,22 @@ where
                 .duration_since(server.connect_time())
                 .as_secs()
                 .to_string(),
+            server
+                .prepared_hit_count
+                .load(Ordering::Relaxed)
+                .to_string(),
+            server
+                .prepared_miss_count
+                .load(Ordering::Relaxed)
+                .to_string(),
+            server
+                .prepared_eviction_count
+                .load(Ordering::Relaxed)
+                .to_string(),
+            server
+                .prepared_cache_size
+                .load(Ordering::Relaxed)
+                .to_string(),
         ];
 
         res.put(data_row(&row));
@@ -764,96 +806,128 @@ where
 }
 
 /// Pause a pool. It won't pass any more queries to the backends.
-async fn pause<T>(stream: &mut T, query: &str) -> Result<(), Error>
+async fn pause<T>(stream: &mut T, tokens: Vec<&str>) -> Result<(), Error>
 where
     T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let parts: Vec<&str> = query.split(",").map(|part| part.trim()).collect();
+    let parts: Vec<&str> = match tokens.len() == 2 {
+        true => tokens[1].split(',').map(|part| part.trim()).collect(),
+        false => Vec::new(),
+    };
 
-    if parts.len() != 2 {
-        error_response(
-            stream,
-            "PAUSE requires a database and a user, e.g. PAUSE my_db,my_user",
-        )
-        .await
-    } else {
-        let database = parts[0];
-        let user = parts[1];
-
-        match get_pool(database, user) {
-            Some(pool) => {
+    match parts.len() {
+        0 => {
+            for (_, pool) in get_all_pools() {
                 pool.pause();
-
-                let mut res = BytesMut::new();
-
-                res.put(command_complete(&format!("PAUSE {},{}", database, user)));
-
-                // ReadyForQuery
-                res.put_u8(b'Z');
-                res.put_i32(5);
-                res.put_u8(b'I');
-
-                write_all_half(stream, &res).await
             }
 
-            None => {
-                error_response(
-                    stream,
-                    &format!(
-                        "No pool configured for database: {}, user: {}",
-                        database, user
-                    ),
-                )
-                .await
+            let mut res = BytesMut::new();
+
+            res.put(command_complete("PAUSE"));
+
+            // ReadyForQuery
+            res.put_u8(b'Z');
+            res.put_i32(5);
+            res.put_u8(b'I');
+
+            write_all_half(stream, &res).await
+        }
+        2 => {
+            let database = parts[0];
+            let user = parts[1];
+
+            match get_pool(database, user) {
+                Some(pool) => {
+                    pool.pause();
+
+                    let mut res = BytesMut::new();
+
+                    res.put(command_complete(&format!("PAUSE {},{}", database, user)));
+
+                    // ReadyForQuery
+                    res.put_u8(b'Z');
+                    res.put_i32(5);
+                    res.put_u8(b'I');
+
+                    write_all_half(stream, &res).await
+                }
+
+                None => {
+                    error_response(
+                        stream,
+                        &format!(
+                            "No pool configured for database: {}, user: {}",
+                            database, user
+                        ),
+                    )
+                    .await
+                }
             }
         }
+        _ => error_response(stream, "usage: PAUSE [db, user]").await,
     }
 }
 
 /// Resume a pool. Queries are allowed again.
-async fn resume<T>(stream: &mut T, query: &str) -> Result<(), Error>
+async fn resume<T>(stream: &mut T, tokens: Vec<&str>) -> Result<(), Error>
 where
     T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let parts: Vec<&str> = query.split(",").map(|part| part.trim()).collect();
+    let parts: Vec<&str> = match tokens.len() == 2 {
+        true => tokens[1].split(',').map(|part| part.trim()).collect(),
+        false => Vec::new(),
+    };
 
-    if parts.len() != 2 {
-        error_response(
-            stream,
-            "RESUME requires a database and a user, e.g. RESUME my_db,my_user",
-        )
-        .await
-    } else {
-        let database = parts[0];
-        let user = parts[1];
-
-        match get_pool(database, user) {
-            Some(pool) => {
+    match parts.len() {
+        0 => {
+            for (_, pool) in get_all_pools() {
                 pool.resume();
-
-                let mut res = BytesMut::new();
-
-                res.put(command_complete(&format!("RESUME {},{}", database, user)));
-
-                // ReadyForQuery
-                res.put_u8(b'Z');
-                res.put_i32(5);
-                res.put_u8(b'I');
-
-                write_all_half(stream, &res).await
             }
 
-            None => {
-                error_response(
-                    stream,
-                    &format!(
-                        "No pool configured for database: {}, user: {}",
-                        database, user
-                    ),
-                )
-                .await
+            let mut res = BytesMut::new();
+
+            res.put(command_complete("RESUME"));
+
+            // ReadyForQuery
+            res.put_u8(b'Z');
+            res.put_i32(5);
+            res.put_u8(b'I');
+
+            write_all_half(stream, &res).await
+        }
+        2 => {
+            let database = parts[0];
+            let user = parts[1];
+
+            match get_pool(database, user) {
+                Some(pool) => {
+                    pool.resume();
+
+                    let mut res = BytesMut::new();
+
+                    res.put(command_complete(&format!("RESUME {},{}", database, user)));
+
+                    // ReadyForQuery
+                    res.put_u8(b'Z');
+                    res.put_i32(5);
+                    res.put_u8(b'I');
+
+                    write_all_half(stream, &res).await
+                }
+
+                None => {
+                    error_response(
+                        stream,
+                        &format!(
+                            "No pool configured for database: {}, user: {}",
+                            database, user
+                        ),
+                    )
+                    .await
+                }
             }
         }
+        _ => error_response(stream, "usage: RESUME [db, user]").await,
     }
 }
 

src/auth_passthrough.rs

@@ -79,6 +79,8 @@ impl AuthPassthrough {
             pool_mode: None,
             server_lifetime: None,
             min_pool_size: None,
+            connect_timeout: None,
+            idle_timeout: None,
         };
 
         let user = &address.username;

src/cmd_args.rs

@@ -0,0 +1,36 @@
+use clap::{Parser, ValueEnum};
+use tracing::Level;
+
+/// PgCat: Nextgen PostgreSQL Pooler
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+pub struct Args {
+    #[arg(default_value_t = String::from("pgcat.toml"), env)]
+    pub config_file: String,
+
+    #[arg(short, long, default_value_t = tracing::Level::INFO, env)]
+    pub log_level: Level,
+
+    #[clap(short='F', long, value_enum, default_value_t=LogFormat::Text, env)]
+    pub log_format: LogFormat,
+
+    #[arg(
+        short,
+        long,
+        default_value_t = false,
+        env,
+        help = "disable colors in the log output"
+    )]
+    pub no_color: bool,
+}
+
+pub fn parse() -> Args {
+    Args::parse()
+}
+
+#[derive(ValueEnum, Clone, Debug)]
+pub enum LogFormat {
+    Text,
+    Structured,
+    Debug,
+}

src/config.rs

@@ -3,11 +3,14 @@ use arc_swap::ArcSwap;
 use log::{error, info};
 use once_cell::sync::Lazy;
 use regex::Regex;
+use serde::{Deserializer, Serializer};
 use serde_derive::{Deserialize, Serialize};
+
 use std::collections::hash_map::DefaultHasher;
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::hash::{Hash, Hasher};
 use std::path::Path;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use tokio::fs::File;
 use tokio::io::AsyncReadExt;
@@ -101,6 +104,9 @@ pub struct Address {
 
     /// Address stats
     pub stats: Arc<AddressStats>,
+
+    /// Number of errors encountered since last successful checkout
+    pub error_count: Arc<AtomicU64>,
 }
 
 impl Default for Address {
@@ -110,14 +116,15 @@ impl Default for Address {
             host: String::from("127.0.0.1"),
             port: 5432,
             shard: 0,
-            address_index: 0,
-            replica_number: 0,
             database: String::from("database"),
             role: Role::Replica,
+            replica_number: 0,
+            address_index: 0,
             username: String::from("username"),
             pool_name: String::from("pool_name"),
             mirrors: Vec::new(),
             stats: Arc::new(AddressStats::default()),
+            error_count: Arc::new(AtomicU64::new(0)),
         }
     }
 }
@@ -182,6 +189,18 @@ impl Address {
             ),
         }
     }
+
+    pub fn error_count(&self) -> u64 {
+        self.error_count.load(Ordering::Relaxed)
+    }
+
+    pub fn increment_error_count(&self) {
+        self.error_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn reset_error_count(&self) {
+        self.error_count.store(0, Ordering::Relaxed);
+    }
 }
 
 /// PostgreSQL user.
@@ -197,6 +216,8 @@ pub struct User {
     pub server_lifetime: Option<u64>,
     #[serde(default)] // 0
     pub statement_timeout: u64,
+    pub connect_timeout: Option<u64>,
+    pub idle_timeout: Option<u64>,
 }
 
 impl Default for User {
@@ -211,24 +232,22 @@ impl Default for User {
             statement_timeout: 0,
             pool_mode: None,
             server_lifetime: None,
+            connect_timeout: None,
+            idle_timeout: None,
         }
     }
 }
 
 impl User {
     fn validate(&self) -> Result<(), Error> {
-        match self.min_pool_size {
-            Some(min_pool_size) => {
-                if min_pool_size > self.pool_size {
-                    error!(
-                        "min_pool_size of {} cannot be larger than pool_size of {}",
-                        min_pool_size, self.pool_size
-                    );
-                    return Err(Error::BadConfig);
-                }
+        if let Some(min_pool_size) = self.min_pool_size {
+            if min_pool_size > self.pool_size {
+                error!(
+                    "min_pool_size of {} cannot be larger than pool_size of {}",
+                    min_pool_size, self.pool_size
+                );
+                return Err(Error::BadConfig);
             }
-
-            None => (),
         };
 
         Ok(())
@@ -261,6 +280,8 @@ pub struct General {
     pub tcp_keepalives_count: u32,
     #[serde(default = "General::default_tcp_keepalives_interval")]
     pub tcp_keepalives_interval: u64,
+    #[serde(default = "General::default_tcp_user_timeout")]
+    pub tcp_user_timeout: u64,
 
     #[serde(default)] // False
     pub log_client_connections: bool,
@@ -292,6 +313,9 @@ pub struct General {
     #[serde(default = "General::default_server_lifetime")]
     pub server_lifetime: u64,
 
+    #[serde(default = "General::default_server_round_robin")] // False
+    pub server_round_robin: bool,
+
     #[serde(default = "General::default_worker_threads")]
     pub worker_threads: usize,
 
@@ -329,7 +353,7 @@ impl General {
     }
 
     pub fn default_server_lifetime() -> u64 {
-        1000 * 60 * 60 * 24 // 24 hours
+        1000 * 60 * 60 // 1 hour
     }
 
     pub fn default_connect_timeout() -> u64 {
@@ -351,8 +375,12 @@ impl General {
         5 // 5 seconds
     }
 
+    pub fn default_tcp_user_timeout() -> u64 {
+        10000 // 10000 milliseconds
+    }
+
     pub fn default_idle_timeout() -> u64 {
-        60000 // 10 minutes
+        600000 // 10 minutes
     }
 
     pub fn default_shutdown_timeout() -> u64 {
@@ -390,6 +418,10 @@ impl General {
     pub fn default_prometheus_exporter_port() -> i16 {
         9930
     }
+
+    pub fn default_server_round_robin() -> bool {
+        true
+    }
 }
 
 impl Default for General {
@@ -401,31 +433,33 @@ impl Default for General {
             prometheus_exporter_port: 9930,
             connect_timeout: General::default_connect_timeout(),
             idle_timeout: General::default_idle_timeout(),
+            tcp_keepalives_idle: Self::default_tcp_keepalives_idle(),
+            tcp_keepalives_count: Self::default_tcp_keepalives_count(),
+            tcp_keepalives_interval: Self::default_tcp_keepalives_interval(),
+            tcp_user_timeout: Self::default_tcp_user_timeout(),
+            log_client_connections: false,
+            log_client_disconnections: false,
+            dns_cache_enabled: false,
+            dns_max_ttl: Self::default_dns_max_ttl(),
             shutdown_timeout: Self::default_shutdown_timeout(),
             healthcheck_timeout: Self::default_healthcheck_timeout(),
             healthcheck_delay: Self::default_healthcheck_delay(),
             ban_time: Self::default_ban_time(),
-            worker_threads: Self::default_worker_threads(),
             idle_client_in_transaction_timeout: Self::default_idle_client_in_transaction_timeout(),
-            tcp_keepalives_idle: Self::default_tcp_keepalives_idle(),
-            tcp_keepalives_count: Self::default_tcp_keepalives_count(),
-            tcp_keepalives_interval: Self::default_tcp_keepalives_interval(),
-            log_client_connections: false,
-            log_client_disconnections: false,
+            server_lifetime: Self::default_server_lifetime(),
+            server_round_robin: Self::default_server_round_robin(),
+            worker_threads: Self::default_worker_threads(),
             autoreload: None,
-            dns_cache_enabled: false,
-            dns_max_ttl: Self::default_dns_max_ttl(),
             tls_certificate: None,
             tls_private_key: None,
             server_tls: false,
             verify_server_certificate: false,
             admin_username: String::from("admin"),
             admin_password: String::from("admin"),
+            validate_config: true,
             auth_query: None,
             auth_query_user: None,
             auth_query_password: None,
-            server_lifetime: 1000 * 3600 * 24, // 24 hours,
-            validate_config: true,
         }
     }
 }
@@ -484,13 +518,23 @@ pub struct Pool {
     #[serde(default)] // False
     pub query_parser_enabled: bool,
 
+    pub query_parser_max_length: Option<usize>,
+
+    #[serde(default)] // False
+    pub query_parser_read_write_splitting: bool,
+
     #[serde(default)] // False
     pub primary_reads_enabled: bool,
 
+    /// Maximum time to allow for establishing a new server connection.
     pub connect_timeout: Option<u64>,
 
+    /// Close idle connections that have been opened for longer than this.
     pub idle_timeout: Option<u64>,
 
+    /// Close server connections that have been opened for longer than this.
+    /// Only applied to idle connections. If the connection is actively used for
+    /// longer than this period, the pool will not interrupt it.
     pub server_lifetime: Option<u64>,
 
     #[serde(default = "Pool::default_sharding_function")]
@@ -503,10 +547,22 @@ pub struct Pool {
     pub shard_id_regex: Option<String>,
     pub regex_search_limit: Option<usize>,
 
+    #[serde(default = "Pool::default_default_shard")]
+    pub default_shard: DefaultShard,
+
     pub auth_query: Option<String>,
     pub auth_query_user: Option<String>,
     pub auth_query_password: Option<String>,
 
+    #[serde(default = "Pool::default_cleanup_server_connections")]
+    pub cleanup_server_connections: bool,
+
+    #[serde(default)] // False
+    pub log_client_parameter_status_changes: bool,
+
+    #[serde(default = "Pool::default_prepared_statements_cache_size")]
+    pub prepared_statements_cache_size: usize,
+
     pub plugins: Option<Plugins>,
     pub shards: BTreeMap<String, Shard>,
     pub users: BTreeMap<String, User>,
@@ -532,6 +588,10 @@ impl Pool {
         PoolMode::Transaction
     }
 
+    pub fn default_default_shard() -> DefaultShard {
+        DefaultShard::default()
+    }
+
     pub fn default_load_balancing_mode() -> LoadBalancingMode {
         LoadBalancingMode::Random
     }
@@ -548,6 +608,14 @@ impl Pool {
         ShardingFunction::PgBigintHash
     }
 
+    pub fn default_cleanup_server_connections() -> bool {
+        true
+    }
+
+    pub fn default_prepared_statements_cache_size() -> usize {
+        0
+    }
+
     pub fn validate(&mut self) -> Result<(), Error> {
         match self.default_role.as_ref() {
             "any" => (),
@@ -588,13 +656,25 @@ impl Pool {
             }
         }
 
+        if self.query_parser_read_write_splitting && !self.query_parser_enabled {
+            error!(
+                "query_parser_read_write_splitting is only valid when query_parser_enabled is true"
+            );
+            return Err(Error::BadConfig);
+        }
+
+        if self.plugins.is_some() && !self.query_parser_enabled {
+            error!("plugins are only valid when query_parser_enabled is true");
+            return Err(Error::BadConfig);
+        }
+
         self.automatic_sharding_key = match &self.automatic_sharding_key {
             Some(key) => {
                 // No quotes in the key so we don't have to compare quoted
                 // to unquoted idents.
-                let key = key.replace("\"", "");
+                let key = key.replace('\"', "");
 
-                if key.split(".").count() != 2 {
+                if key.split('.').count() != 2 {
                     error!(
                         "automatic_sharding_key '{}' must be fully qualified, e.g. t.{}`",
                         key, key
@@ -607,7 +687,14 @@ impl Pool {
             None => None,
         };
 
-        for (_, user) in &self.users {
+        if let DefaultShard::Shard(shard_number) = self.default_shard {
+            if shard_number >= self.shards.len() {
+                error!("Invalid shard {:?}", shard_number);
+                return Err(Error::BadConfig);
+            }
+        }
+
+        for user in self.users.values() {
             user.validate()?;
         }
 
@@ -620,23 +707,29 @@ impl Default for Pool {
         Pool {
             pool_mode: Self::default_pool_mode(),
             load_balancing_mode: Self::default_load_balancing_mode(),
-            shards: BTreeMap::from([(String::from("1"), Shard::default())]),
-            users: BTreeMap::default(),
             default_role: String::from("any"),
             query_parser_enabled: false,
+            query_parser_max_length: None,
+            query_parser_read_write_splitting: false,
             primary_reads_enabled: false,
-            sharding_function: ShardingFunction::PgBigintHash,
-            automatic_sharding_key: None,
             connect_timeout: None,
             idle_timeout: None,
+            server_lifetime: None,
+            sharding_function: ShardingFunction::PgBigintHash,
+            automatic_sharding_key: None,
             sharding_key_regex: None,
             shard_id_regex: None,
             regex_search_limit: Some(1000),
+            default_shard: Self::default_default_shard(),
             auth_query: None,
             auth_query_user: None,
             auth_query_password: None,
-            server_lifetime: None,
+            cleanup_server_connections: true,
+            log_client_parameter_status_changes: false,
+            prepared_statements_cache_size: Self::default_prepared_statements_cache_size(),
             plugins: None,
+            shards: BTreeMap::from([(String::from("1"), Shard::default())]),
+            users: BTreeMap::default(),
         }
     }
 }
@@ -648,6 +741,50 @@ pub struct ServerConfig {
     pub role: Role,
 }
 
+// No Shard Specified handling.
+#[derive(Debug, PartialEq, Clone, Eq, Hash, Copy)]
+pub enum DefaultShard {
+    Shard(usize),
+    Random,
+    RandomHealthy,
+}
+impl Default for DefaultShard {
+    fn default() -> Self {
+        DefaultShard::Shard(0)
+    }
+}
+impl serde::Serialize for DefaultShard {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        match self {
+            DefaultShard::Shard(shard) => {
+                serializer.serialize_str(&format!("shard_{}", &shard.to_string()))
+            }
+            DefaultShard::Random => serializer.serialize_str("random"),
+            DefaultShard::RandomHealthy => serializer.serialize_str("random_healthy"),
+        }
+    }
+}
+impl<'de> serde::Deserialize<'de> for DefaultShard {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        if let Some(s) = s.strip_prefix("shard_") {
+            let shard = s.parse::<usize>().map_err(serde::de::Error::custom)?;
+            return Ok(DefaultShard::Shard(shard));
+        }
+
+        match s.as_str() {
+            "random" => Ok(DefaultShard::Random),
+            "random_healthy" => Ok(DefaultShard::RandomHealthy),
+            _ => Err(serde::de::Error::custom(
+                "invalid value for no_shard_specified_behavior",
+            )),
+        }
+    }
+}
+
 #[derive(Clone, PartialEq, Serialize, Deserialize, Debug, Hash, Eq)]
 pub struct MirrorServerConfig {
     pub host: String,
@@ -704,13 +841,13 @@ impl Shard {
 impl Default for Shard {
     fn default() -> Shard {
         Shard {
+            database: String::from("postgres"),
+            mirrors: None,
             servers: vec![ServerConfig {
                 host: String::from("localhost"),
                 port: 5432,
                 role: Role::Primary,
             }],
-            mirrors: None,
-            database: String::from("postgres"),
         }
     }
 }
@@ -723,15 +860,26 @@ pub struct Plugins {
     pub prewarmer: Option<Prewarmer>,
 }
 
+pub trait Plugin {
+    fn is_enabled(&self) -> bool;
+}
+
 impl std::fmt::Display for Plugins {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        fn is_enabled<T: Plugin>(arg: Option<&T>) -> bool {
+            if let Some(arg) = arg {
+                arg.is_enabled()
+            } else {
+                false
+            }
+        }
         write!(
             f,
             "interceptor: {}, table_access: {}, query_logger: {}, prewarmer: {}",
-            self.intercept.is_some(),
-            self.table_access.is_some(),
-            self.query_logger.is_some(),
-            self.prewarmer.is_some(),
+            is_enabled(self.intercept.as_ref()),
+            is_enabled(self.table_access.as_ref()),
+            is_enabled(self.query_logger.as_ref()),
+            is_enabled(self.prewarmer.as_ref()),
         )
     }
 }
@@ -742,23 +890,47 @@ pub struct Intercept {
     pub queries: BTreeMap<String, Query>,
 }
 
+impl Plugin for Intercept {
+    fn is_enabled(&self) -> bool {
+        self.enabled
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
 pub struct TableAccess {
     pub enabled: bool,
     pub tables: Vec<String>,
 }
 
+impl Plugin for TableAccess {
+    fn is_enabled(&self) -> bool {
+        self.enabled
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
 pub struct QueryLogger {
     pub enabled: bool,
 }
 
+impl Plugin for QueryLogger {
+    fn is_enabled(&self) -> bool {
+        self.enabled
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
 pub struct Prewarmer {
     pub enabled: bool,
     pub queries: Vec<String>,
 }
 
+impl Plugin for Prewarmer {
+    fn is_enabled(&self) -> bool {
+        self.enabled
+    }
+}
+
 impl Intercept {
     pub fn substitute(&mut self, db: &str, user: &str) {
         for (_, query) in self.queries.iter_mut() {
@@ -776,6 +948,7 @@ pub struct Query {
 }
 
 impl Query {
+    #[allow(clippy::needless_range_loop)]
     pub fn substitute(&mut self, db: &str, user: &str) {
         for col in self.result.iter_mut() {
             for i in 0..col.len() {
@@ -845,8 +1018,8 @@ impl Default for Config {
         Config {
             path: Self::default_path(),
             general: General::default(),
-            pools: HashMap::default(),
             plugins: None,
+            pools: HashMap::default(),
         }
     }
 }
@@ -874,6 +1047,17 @@ impl From<&Config> for std::collections::HashMap<String, String> {
                         format!("pools.{}.query_parser_enabled", pool_name),
                         pool.query_parser_enabled.to_string(),
                     ),
+                    (
+                        format!("pools.{}.query_parser_max_length", pool_name),
+                        match pool.query_parser_max_length {
+                            Some(max_length) => max_length.to_string(),
+                            None => String::from("unlimited"),
+                        },
+                    ),
+                    (
+                        format!("pools.{}.query_parser_read_write_splitting", pool_name),
+                        pool.query_parser_read_write_splitting.to_string(),
+                    ),
                     (
                         format!("pools.{}.default_role", pool_name),
                         pool.default_role.clone(),
@@ -889,8 +1073,8 @@ impl From<&Config> for std::collections::HashMap<String, String> {
                     (
                         format!("pools.{:?}.users", pool_name),
                         pool.users
-                            .iter()
-                            .map(|(_username, user)| &user.username)
+                            .values()
+                            .map(|user| &user.username)
                             .cloned()
                             .collect::<Vec<String>>()
                             .join(", "),
@@ -944,6 +1128,7 @@ impl From<&Config> for std::collections::HashMap<String, String> {
 impl Config {
     /// Print current configuration.
     pub fn show(&self) {
+        info!("Config path: {}", self.path);
         info!("Ban time: {}s", self.general.ban_time);
         info!(
             "Idle client in transaction timeout: {}ms",
@@ -970,17 +1155,14 @@ impl Config {
             "Default max server lifetime: {}ms",
             self.general.server_lifetime
         );
+        info!("Sever round robin: {}", self.general.server_round_robin);
         match self.general.tls_certificate.clone() {
             Some(tls_certificate) => {
                 info!("TLS certificate: {}", tls_certificate);
 
-                match self.general.tls_private_key.clone() {
-                    Some(tls_private_key) => {
-                        info!("TLS private key: {}", tls_private_key);
-                        info!("TLS support is enabled");
-                    }
-
-                    None => (),
+                if let Some(tls_private_key) = self.general.tls_private_key.clone() {
+                    info!("TLS private key: {}", tls_private_key);
+                    info!("TLS support is enabled");
                 }
             }
 
@@ -1008,8 +1190,8 @@ impl Config {
                 pool_name,
                 pool_config
                     .users
-                    .iter()
-                    .map(|(_, user_cfg)| user_cfg.pool_size)
+                    .values()
+                    .map(|user_cfg| user_cfg.pool_size)
                     .sum::<u32>()
                     .to_string()
             );
@@ -1048,6 +1230,15 @@ impl Config {
                 "[pool: {}] Query router: {}",
                 pool_name, pool_config.query_parser_enabled
             );
+
+            info!(
+                "[pool: {}] Query parser max length: {:?}",
+                pool_name, pool_config.query_parser_max_length
+            );
+            info!(
+                "[pool: {}] Infer role from query: {}",
+                pool_name, pool_config.query_parser_read_write_splitting
+            );
             info!(
                 "[pool: {}] Number of shards: {}",
                 pool_name,
@@ -1066,6 +1257,18 @@ impl Config {
                     None => "default".to_string(),
                 }
             );
+            info!(
+                "[pool: {}] Cleanup server connections: {}",
+                pool_name, pool_config.cleanup_server_connections
+            );
+            info!(
+                "[pool: {}] Log client parameter status changes: {}",
+                pool_name, pool_config.log_client_parameter_status_changes
+            );
+            info!(
+                "[pool: {}] Prepared statements server cache size: {}",
+                pool_name, pool_config.prepared_statements_cache_size
+            );
             info!(
                 "[pool: {}] Plugins: {}",
                 pool_name,
@@ -1108,6 +1311,24 @@ impl Config {
                         None => "default".to_string(),
                     }
                 );
+                info!(
+                    "[pool: {}][user: {}] Connection timeout: {}",
+                    pool_name,
+                    user.1.username,
+                    match user.1.connect_timeout {
+                        Some(connect_timeout) => format!("{}ms", connect_timeout),
+                        None => "not set".to_string(),
+                    }
+                );
+                info!(
+                    "[pool: {}][user: {}] Idle timeout: {}",
+                    pool_name,
+                    user.1.username,
+                    match user.1.idle_timeout {
+                        Some(idle_timeout) => format!("{}ms", idle_timeout),
+                        None => "not set".to_string(),
+                    }
+                );
             }
         }
     }
@@ -1162,34 +1383,31 @@ impl Config {
         }
 
         // Validate TLS!
-        match self.general.tls_certificate.clone() {
-            Some(tls_certificate) => {
-                match load_certs(Path::new(&tls_certificate)) {
-                    Ok(_) => {
-                        // Cert is okay, but what about the private key?
-                        match self.general.tls_private_key.clone() {
-                            Some(tls_private_key) => match load_keys(Path::new(&tls_private_key)) {
-                                Ok(_) => (),
-                                Err(err) => {
-                                    error!("tls_private_key is incorrectly configured: {:?}", err);
-                                    return Err(Error::BadConfig);
-                                }
-                            },
-
-                            None => {
-                                error!("tls_certificate is set, but the tls_private_key is not");
+        if let Some(tls_certificate) = self.general.tls_certificate.clone() {
+            match load_certs(Path::new(&tls_certificate)) {
+                Ok(_) => {
+                    // Cert is okay, but what about the private key?
+                    match self.general.tls_private_key.clone() {
+                        Some(tls_private_key) => match load_keys(Path::new(&tls_private_key)) {
+                            Ok(_) => (),
+                            Err(err) => {
+                                error!("tls_private_key is incorrectly configured: {:?}", err);
                                 return Err(Error::BadConfig);
                             }
-                        };
-                    }
+                        },
 
-                    Err(err) => {
-                        error!("tls_certificate is incorrectly configured: {:?}", err);
-                        return Err(Error::BadConfig);
-                    }
+                        None => {
+                            error!("tls_certificate is set, but the tls_private_key is not");
+                            return Err(Error::BadConfig);
+                        }
+                    };
+                }
+
+                Err(err) => {
+                    error!("tls_certificate is incorrectly configured: {:?}", err);
+                    return Err(Error::BadConfig);
                 }
             }
-            None => (),
         };
 
         for pool in self.pools.values_mut() {
@@ -1208,9 +1426,7 @@ pub fn get_config() -> Config {
 }
 
 pub fn get_idle_client_in_transaction_timeout() -> u64 {
-    (*(*CONFIG.load()))
-        .general
-        .idle_client_in_transaction_timeout
+    CONFIG.load().general.idle_client_in_transaction_timeout
 }
 
 /// Parse the configuration file located at the path.

src/errors.rs

@@ -12,6 +12,7 @@ pub enum Error {
     ProtocolSyncError(String),
     BadQuery(String),
     ServerError,
+    ServerMessageParserError(String),
     ServerStartupError(String, ServerIdentifier),
     ServerAuthError(String, ServerIdentifier),
     BadConfig,
@@ -26,6 +27,9 @@ pub enum Error {
     AuthPassthroughError(String),
     UnsupportedStatement,
     QueryRouterParserError(String),
+    QueryRouterError(String),
+    InvalidShardId(usize),
+    PreparedStatementError,
 }
 
 #[derive(Clone, PartialEq, Debug)]
@@ -121,3 +125,9 @@ impl std::fmt::Display for Error {
         }
     }
 }
+
+impl From<std::ffi::NulError> for Error {
+    fn from(err: std::ffi::NulError) -> Self {
+        Error::QueryRouterError(err.to_string())
+    }
+}

src/lib.rs

@@ -1,13 +1,14 @@
 pub mod admin;
 pub mod auth_passthrough;
 pub mod client;
+pub mod cmd_args;
 pub mod config;
 pub mod constants;
 pub mod dns_cache;
 pub mod errors;
+pub mod logger;
 pub mod messages;
 pub mod mirrors;
-pub mod multi_logger;
 pub mod plugins;
 pub mod pool;
 pub mod prometheus;

src/logger.rs

@@ -0,0 +1,20 @@
+use crate::cmd_args::{Args, LogFormat};
+use tracing_subscriber;
+use tracing_subscriber::EnvFilter;
+
+pub fn init(args: &Args) {
+    // Iniitalize a default filter, and then override the builtin default "warning" with our
+    // commandline, (default: "info")
+    let filter = EnvFilter::from_default_env().add_directive(args.log_level.into());
+
+    let trace_sub = tracing_subscriber::fmt()
+        .with_thread_ids(true)
+        .with_env_filter(filter)
+        .with_ansi(!args.no_color);
+
+    match args.log_format {
+        LogFormat::Structured => trace_sub.json().init(),
+        LogFormat::Debug => trace_sub.pretty().init(),
+        _ => trace_sub.init(),
+    };
+}

src/main.rs

@@ -23,7 +23,6 @@ extern crate arc_swap;
 extern crate async_trait;
 extern crate bb8;
 extern crate bytes;
-extern crate env_logger;
 extern crate exitcode;
 extern crate log;
 extern crate md5;
@@ -61,15 +60,18 @@ use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::broadcast;
 
+use pgcat::cmd_args;
 use pgcat::config::{get_config, reload_config, VERSION};
 use pgcat::dns_cache;
+use pgcat::logger;
 use pgcat::messages::configure_socket;
 use pgcat::pool::{ClientServerMap, ConnectionPool};
 use pgcat::prometheus::start_metric_server;
 use pgcat::stats::{Collector, Reporter, REPORTER};
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    pgcat::multi_logger::MultiLogger::init().unwrap();
+    let args = cmd_args::parse();
+    logger::init(&args);
 
     info!("Welcome to PgCat! Meow. (Version {})", VERSION);
 
@@ -78,20 +80,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         std::process::exit(exitcode::CONFIG);
     }
 
-    let args = std::env::args().collect::<Vec<String>>();
-
-    let config_file = if args.len() == 2 {
-        args[1].to_string()
-    } else {
-        String::from("pgcat.toml")
-    };
-
     // Create a transient runtime for loading the config for the first time.
     {
         let runtime = Builder::new_multi_thread().worker_threads(1).build()?;
 
         runtime.block_on(async {
-            match pgcat::config::parse(&config_file).await {
+            match pgcat::config::parse(args.config_file.as_str()).await {
                 Ok(_) => (),
                 Err(err) => {
                     error!("Config parse error: {:?}", err);

src/mirrors.rs

@@ -7,8 +7,7 @@ use bytes::{Bytes, BytesMut};
 use parking_lot::RwLock;
 
 use crate::config::{get_config, Address, Role, User};
-use crate::pool::{ClientServerMap, PoolIdentifier, ServerPool};
-use crate::stats::PoolStats;
+use crate::pool::{ClientServerMap, ServerPool};
 use log::{error, info, trace, warn};
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 
@@ -24,26 +23,27 @@ impl MirroredClient {
     async fn create_pool(&self) -> Pool<ServerPool> {
         let config = get_config();
         let default = std::time::Duration::from_millis(10_000).as_millis() as u64;
-        let (connection_timeout, idle_timeout, cfg) =
+        let (connection_timeout, idle_timeout, _cfg, prepared_statement_cache_size) =
             match config.pools.get(&self.address.pool_name) {
                 Some(cfg) => (
                     cfg.connect_timeout.unwrap_or(default),
                     cfg.idle_timeout.unwrap_or(default),
                     cfg.clone(),
+                    cfg.prepared_statements_cache_size,
                 ),
-                None => (default, default, crate::config::Pool::default()),
+                None => (default, default, crate::config::Pool::default(), 0),
             };
 
-        let identifier = PoolIdentifier::new(&self.database, &self.user.username);
-
         let manager = ServerPool::new(
             self.address.clone(),
             self.user.clone(),
             self.database.as_str(),
             ClientServerMap::default(),
-            Arc::new(PoolStats::new(identifier, cfg.clone())),
             Arc::new(RwLock::new(None)),
             None,
+            true,
+            false,
+            prepared_statement_cache_size,
         );
 
         Pool::builder()
@@ -81,7 +81,7 @@ impl MirroredClient {
                     }
 
                     // Incoming data from server (we read to clear the socket buffer and discard the data)
-                    recv_result = server.recv() => {
+                    recv_result = server.recv(None) => {
                         match recv_result {
                             Ok(message) => trace!("Received from mirror: {} {:?}", String::from_utf8_lossy(&message[..]), address.clone()),
                             Err(err) => {
@@ -139,18 +139,18 @@ impl MirroringManager {
                 bytes_rx,
                 disconnect_rx: exit_rx,
             };
-            exit_senders.push(exit_tx.clone());
-            byte_senders.push(bytes_tx.clone());
+            exit_senders.push(exit_tx);
+            byte_senders.push(bytes_tx);
             client.start();
         });
 
         Self {
-            byte_senders: byte_senders,
+            byte_senders,
             disconnect_senders: exit_senders,
         }
     }
 
-    pub fn send(self: &mut Self, bytes: &BytesMut) {
+    pub fn send(&mut self, bytes: &BytesMut) {
         // We want to avoid performing an allocation if we won't be able to send the message
         // There is a possibility of a race here where we check the capacity and then the channel is
         // closed or the capacity is reduced to 0, but mirroring is best effort anyway
@@ -172,7 +172,7 @@ impl MirroringManager {
         });
     }
 
-    pub fn disconnect(self: &mut Self) {
+    pub fn disconnect(&mut self) {
         self.disconnect_senders
             .iter_mut()
             .for_each(|sender| match sender.try_send(()) {

src/multi_logger.rs

@@ -1,80 +0,0 @@
-use log::{Level, Log, Metadata, Record, SetLoggerError};
-
-// This is a special kind of logger that allows sending logs to different
-// targets depending on the log level.
-//
-// By default, if nothing is set, it acts as a regular env_log logger,
-// it sends everything to standard error.
-//
-// If the Env variable `STDOUT_LOG` is defined, it will be used for
-// configuring the standard out logger.
-//
-// The behavior is:
-//   - If it is an error, the message is written to standard error.
-//   - If it is not, and it matches the log level of the standard output logger (`STDOUT_LOG` env var), it will be send to standard output.
-//   - If the above is not true, it is sent to the stderr logger that will log it or not depending on the value
-//     of the RUST_LOG env var.
-//
-// So to summarize, if no `STDOUT_LOG` env var is present, the logger is the default logger. If `STDOUT_LOG` is set, everything
-// but errors, that matches the log level set in the `STDOUT_LOG` env var is sent to stdout. You can have also some esoteric configuration
-// where you set `RUST_LOG=debug` and `STDOUT_LOG=info`, in here, errors will go to stderr, warns and infos to stdout and debugs to stderr.
-//
-pub struct MultiLogger {
-    stderr_logger: env_logger::Logger,
-    stdout_logger: env_logger::Logger,
-}
-
-impl MultiLogger {
-    fn new() -> Self {
-        let stderr_logger = env_logger::builder().format_timestamp_micros().build();
-        let stdout_logger = env_logger::Builder::from_env("STDOUT_LOG")
-            .format_timestamp_micros()
-            .target(env_logger::Target::Stdout)
-            .build();
-
-        Self {
-            stderr_logger,
-            stdout_logger,
-        }
-    }
-
-    pub fn init() -> Result<(), SetLoggerError> {
-        let logger = Self::new();
-
-        log::set_max_level(logger.stderr_logger.filter());
-        log::set_boxed_logger(Box::new(logger))
-    }
-}
-
-impl Log for MultiLogger {
-    fn enabled(&self, metadata: &Metadata) -> bool {
-        self.stderr_logger.enabled(metadata) && self.stdout_logger.enabled(metadata)
-    }
-
-    fn log(&self, record: &Record) {
-        if record.level() == Level::Error {
-            self.stderr_logger.log(record);
-        } else {
-            if self.stdout_logger.matches(record) {
-                self.stdout_logger.log(record);
-            } else {
-                self.stderr_logger.log(record);
-            }
-        }
-    }
-
-    fn flush(&self) {
-        self.stderr_logger.flush();
-        self.stdout_logger.flush();
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-
-    #[test]
-    fn test_init() {
-        MultiLogger::init().unwrap();
-    }
-}

src/plugins/intercept.rs

@@ -92,7 +92,7 @@ impl<'a> Plugin for Intercept<'a> {
                             .map(|s| {
                                 let s = s.as_str().to_string();
 
-                                if s == "" {
+                                if s.is_empty() {
                                     None
                                 } else {
                                     Some(s)

src/plugins/mod.rs

@@ -33,6 +33,7 @@ pub enum PluginOutput {
 #[async_trait]
 pub trait Plugin {
     // Run before the query is sent to the server.
+    #[allow(clippy::ptr_arg)]
     async fn run(
         &mut self,
         query_router: &QueryRouter,

src/plugins/prewarmer.rs

@@ -20,7 +20,7 @@ impl<'a> Prewarmer<'a> {
                 self.server.address(),
                 query
             );
-            self.server.query(&query).await?;
+            self.server.query(query).await?;
         }
 
         Ok(())

src/plugins/query_logger.rs

@@ -31,7 +31,7 @@ impl<'a> Plugin for QueryLogger<'a> {
             .map(|q| q.to_string())
             .collect::<Vec<String>>()
             .join("; ");
-        info!("[pool: {}][user: {}] {}", self.user, self.db, query);
+        info!("[pool: {}][user: {}] {}", self.db, self.user, query);
 
         Ok(PluginOutput::Allow)
     }

src/plugins/table_access.rs

@@ -34,7 +34,7 @@ impl<'a> Plugin for TableAccess<'a> {
 
         visit_relations(ast, |relation| {
             let relation = relation.to_string();
-            let parts = relation.split(".").collect::<Vec<&str>>();
+            let parts = relation.split('.').collect::<Vec<&str>>();
             let table_name = parts.last().unwrap();
 
             if self.tables.contains(&table_name.to_string()) {

src/pool.rs

@@ -1,15 +1,18 @@
 use arc_swap::ArcSwap;
 use async_trait::async_trait;
-use bb8::{ManageConnection, Pool, PooledConnection};
-use bytes::{BufMut, BytesMut};
+use bb8::{ManageConnection, Pool, PooledConnection, QueueStrategy};
 use chrono::naive::NaiveDateTime;
 use log::{debug, error, info, warn};
+use lru::LruCache;
 use once_cell::sync::Lazy;
 use parking_lot::{Mutex, RwLock};
 use rand::seq::SliceRandom;
 use rand::thread_rng;
 use regex::Regex;
 use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
+use std::num::NonZeroUsize;
+use std::sync::atomic::AtomicU64;
 use std::sync::{
     atomic::{AtomicBool, Ordering},
     Arc,
@@ -18,15 +21,16 @@ use std::time::Instant;
 use tokio::sync::Notify;
 
 use crate::config::{
-    get_config, Address, General, LoadBalancingMode, Plugins, PoolMode, Role, User,
+    get_config, Address, DefaultShard, General, LoadBalancingMode, Plugins, PoolMode, Role, User,
 };
 use crate::errors::Error;
 
 use crate::auth_passthrough::AuthPassthrough;
+use crate::messages::Parse;
 use crate::plugins::prewarmer;
-use crate::server::Server;
+use crate::server::{Server, ServerParameters};
 use crate::sharding::ShardingFunction;
-use crate::stats::{AddressStats, ClientStats, PoolStats, ServerStats};
+use crate::stats::{AddressStats, ClientStats, ServerStats};
 
 pub type ProcessId = i32;
 pub type SecretKey = i32;
@@ -53,6 +57,57 @@ pub enum BanReason {
     AdminBan(i64),
 }
 
+pub type PreparedStatementCacheType = Arc<Mutex<PreparedStatementCache>>;
+
+// TODO: Add stats the this cache
+// TODO: Add application name to the cache value to help identify which application is using the cache
+// TODO: Create admin command to show which statements are in the cache
+#[derive(Debug)]
+pub struct PreparedStatementCache {
+    cache: LruCache<u64, Arc<Parse>>,
+}
+
+impl PreparedStatementCache {
+    pub fn new(mut size: usize) -> Self {
+        // Cannot be zeros
+        if size == 0 {
+            size = 1;
+        }
+
+        PreparedStatementCache {
+            cache: LruCache::new(NonZeroUsize::new(size).unwrap()),
+        }
+    }
+
+    /// Adds the prepared statement to the cache if it doesn't exist with a new name
+    /// if it already exists will give you the existing parse
+    ///
+    /// Pass the hash to this so that we can do the compute before acquiring the lock
+    pub fn get_or_insert(&mut self, parse: &Parse, hash: u64) -> Arc<Parse> {
+        match self.cache.get(&hash) {
+            Some(rewritten_parse) => rewritten_parse.clone(),
+            None => {
+                let new_parse = Arc::new(parse.clone().rewrite());
+                let evicted = self.cache.push(hash, new_parse.clone());
+
+                if let Some((_, evicted_parse)) = evicted {
+                    debug!(
+                        "Evicted prepared statement {} from cache",
+                        evicted_parse.name
+                    );
+                }
+
+                new_parse
+            }
+        }
+    }
+
+    /// Marks the hash as most recently used if it exists
+    pub fn promote(&mut self, hash: &u64) {
+        self.cache.promote(hash);
+    }
+}
+
 /// An identifier for a PgCat pool,
 /// a database visible to clients.
 #[derive(Hash, Debug, Clone, PartialEq, Eq, Default)]
@@ -76,6 +131,12 @@ impl PoolIdentifier {
     }
 }
 
+impl Display for PoolIdentifier {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}@{}", self.user, self.db)
+    }
+}
+
 impl From<&Address> for PoolIdentifier {
     fn from(address: &Address) -> PoolIdentifier {
         PoolIdentifier::new(&address.database, &address.username)
@@ -104,6 +165,12 @@ pub struct PoolSettings {
     // Enable/disable query parser.
     pub query_parser_enabled: bool,
 
+    // Max length of query the parser will parse.
+    pub query_parser_max_length: Option<usize>,
+
+    // Infer role
+    pub query_parser_read_write_splitting: bool,
+
     // Read from the primary as well or not.
     pub primary_reads_enabled: bool,
 
@@ -128,6 +195,9 @@ pub struct PoolSettings {
     // Regex for searching for the shard id in SQL statements
     pub shard_id_regex: Option<Regex>,
 
+    // What to do when no shard is selected in a sharded system
+    pub default_shard: DefaultShard,
+
     // Limit how much of each query is searched for a potential shard regex match
     pub regex_search_limit: usize,
 
@@ -150,6 +220,8 @@ impl Default for PoolSettings {
             db: String::default(),
             default_role: None,
             query_parser_enabled: false,
+            query_parser_max_length: None,
+            query_parser_read_write_splitting: false,
             primary_reads_enabled: true,
             sharding_function: ShardingFunction::PgBigintHash,
             automatic_sharding_key: None,
@@ -159,6 +231,7 @@ impl Default for PoolSettings {
             sharding_key_regex: None,
             shard_id_regex: None,
             regex_search_limit: 1000,
+            default_shard: DefaultShard::Shard(0),
             auth_query: None,
             auth_query_user: None,
             auth_query_password: None,
@@ -171,23 +244,23 @@ impl Default for PoolSettings {
 #[derive(Clone, Debug, Default)]
 pub struct ConnectionPool {
     /// The pools handled internally by bb8.
-    databases: Vec<Vec<Pool<ServerPool>>>,
+    databases: Arc<Vec<Vec<Pool<ServerPool>>>>,
 
     /// The addresses (host, port, role) to handle
     /// failover and load balancing deterministically.
-    addresses: Vec<Vec<Address>>,
+    addresses: Arc<Vec<Vec<Address>>>,
 
     /// List of banned addresses (see above)
     /// that should not be queried.
     banlist: BanList,
 
-    /// The server information (K messages) have to be passed to the
+    /// The server information has to be passed to the
     /// clients on startup. We pre-connect to all shards and replicas
-    /// on pool creation and save the K messages here.
-    server_info: Arc<RwLock<BytesMut>>,
+    /// on pool creation and save the startup parameters here.
+    original_server_parameters: Arc<RwLock<ServerParameters>>,
 
     /// Pool configuration.
-    pub settings: PoolSettings,
+    pub settings: Arc<PoolSettings>,
 
     /// If not validated, we need to double check the pool is available before allowing a client
     /// to use it.
@@ -202,11 +275,11 @@ pub struct ConnectionPool {
     paused: Arc<AtomicBool>,
     paused_waiter: Arc<Notify>,
 
-    /// Statistics.
-    pub stats: Arc<PoolStats>,
-
     /// AuthInfo
     pub auth_hash: Arc<RwLock<Option<String>>>,
+
+    /// Cache
+    pub prepared_statement_cache: Option<PreparedStatementCacheType>,
 }
 
 impl ConnectionPool {
@@ -225,20 +298,17 @@ impl ConnectionPool {
                 let old_pool_ref = get_pool(pool_name, &user.username);
                 let identifier = PoolIdentifier::new(pool_name, &user.username);
 
-                match old_pool_ref {
-                    Some(pool) => {
-                        // If the pool hasn't changed, get existing reference and insert it into the new_pools.
-                        // We replace all pools at the end, but if the reference is kept, the pool won't get re-created (bb8).
-                        if pool.config_hash == new_pool_hash_value {
-                            info!(
-                                "[pool: {}][user: {}] has not changed",
-                                pool_name, user.username
-                            );
-                            new_pools.insert(identifier.clone(), pool.clone());
-                            continue;
-                        }
+                if let Some(pool) = old_pool_ref {
+                    // If the pool hasn't changed, get existing reference and insert it into the new_pools.
+                    // We replace all pools at the end, but if the reference is kept, the pool won't get re-created (bb8).
+                    if pool.config_hash == new_pool_hash_value {
+                        info!(
+                            "[pool: {}][user: {}] has not changed",
+                            pool_name, user.username
+                        );
+                        new_pools.insert(identifier.clone(), pool.clone());
+                        continue;
                     }
-                    None => (),
                 }
 
                 info!(
@@ -254,10 +324,6 @@ impl ConnectionPool {
                     .clone()
                     .into_keys()
                     .collect::<Vec<String>>();
-                let pool_stats = Arc::new(PoolStats::new(identifier, pool_config.clone()));
-
-                // Allow the pool to be seen in statistics
-                pool_stats.register(pool_stats.clone());
 
                 // Sort by shard number to ensure consistency.
                 shard_ids.sort_by_key(|k| k.parse::<i64>().unwrap());
@@ -292,6 +358,7 @@ impl ConnectionPool {
                                     pool_name: pool_name.clone(),
                                     mirrors: vec![],
                                     stats: Arc::new(AddressStats::default()),
+                                    error_count: Arc::new(AtomicU64::new(0)),
                                 });
                                 address_id += 1;
                             }
@@ -310,6 +377,7 @@ impl ConnectionPool {
                             pool_name: pool_name.clone(),
                             mirrors: mirror_addresses,
                             stats: Arc::new(AddressStats::default()),
+                            error_count: Arc::new(AtomicU64::new(0)),
                         };
 
                         address_id += 1;
@@ -358,22 +426,30 @@ impl ConnectionPool {
                             user.clone(),
                             &shard.database,
                             client_server_map.clone(),
-                            pool_stats.clone(),
                             pool_auth_hash.clone(),
                             match pool_config.plugins {
                                 Some(ref plugins) => Some(plugins.clone()),
                                 None => config.plugins.clone(),
                             },
+                            pool_config.cleanup_server_connections,
+                            pool_config.log_client_parameter_status_changes,
+                            pool_config.prepared_statements_cache_size,
                         );
 
-                        let connect_timeout = match pool_config.connect_timeout {
+                        let connect_timeout = match user.connect_timeout {
                             Some(connect_timeout) => connect_timeout,
-                            None => config.general.connect_timeout,
+                            None => match pool_config.connect_timeout {
+                                Some(connect_timeout) => connect_timeout,
+                                None => config.general.connect_timeout,
+                            },
                         };
 
-                        let idle_timeout = match pool_config.idle_timeout {
+                        let idle_timeout = match user.idle_timeout {
                             Some(idle_timeout) => idle_timeout,
-                            None => config.general.idle_timeout,
+                            None => match pool_config.idle_timeout {
+                                Some(idle_timeout) => idle_timeout,
+                                None => config.general.idle_timeout,
+                            },
                         };
 
                         let server_lifetime = match user.server_lifetime {
@@ -384,11 +460,16 @@ impl ConnectionPool {
                             },
                         };
 
-                        let reaper_rate = *vec![idle_timeout, server_lifetime, POOL_REAPER_RATE]
+                        let reaper_rate = *[idle_timeout, server_lifetime, POOL_REAPER_RATE]
                             .iter()
                             .min()
                             .unwrap();
 
+                        let queue_strategy = match config.general.server_round_robin {
+                            true => QueueStrategy::Fifo,
+                            false => QueueStrategy::Lifo,
+                        };
+
                         debug!(
                             "[pool: {}][user: {}] Pool reaper rate: {}ms",
                             pool_name, user.username, reaper_rate
@@ -401,6 +482,7 @@ impl ConnectionPool {
                             .idle_timeout(Some(std::time::Duration::from_millis(idle_timeout)))
                             .max_lifetime(Some(std::time::Duration::from_millis(server_lifetime)))
                             .reaper_rate(std::time::Duration::from_millis(reaper_rate))
+                            .queue_strategy(queue_strategy)
                             .test_on_check_out(false);
 
                         let pool = if config.general.validate_config {
@@ -427,14 +509,13 @@ impl ConnectionPool {
                 }
 
                 let pool = ConnectionPool {
-                    databases: shards,
-                    stats: pool_stats,
-                    addresses,
+                    databases: Arc::new(shards),
+                    addresses: Arc::new(addresses),
                     banlist: Arc::new(RwLock::new(banlist)),
                     config_hash: new_pool_hash_value,
-                    server_info: Arc::new(RwLock::new(BytesMut::new())),
+                    original_server_parameters: Arc::new(RwLock::new(ServerParameters::new())),
                     auth_hash: pool_auth_hash,
-                    settings: PoolSettings {
+                    settings: Arc::new(PoolSettings {
                         pool_mode: match user.pool_mode {
                             Some(pool_mode) => pool_mode,
                             None => pool_config.pool_mode,
@@ -451,6 +532,9 @@ impl ConnectionPool {
                             _ => unreachable!(),
                         },
                         query_parser_enabled: pool_config.query_parser_enabled,
+                        query_parser_max_length: pool_config.query_parser_max_length,
+                        query_parser_read_write_splitting: pool_config
+                            .query_parser_read_write_splitting,
                         primary_reads_enabled: pool_config.primary_reads_enabled,
                         sharding_function: pool_config.sharding_function,
                         automatic_sharding_key: pool_config.automatic_sharding_key.clone(),
@@ -466,6 +550,7 @@ impl ConnectionPool {
                             .clone()
                             .map(|regex| Regex::new(regex.as_str()).unwrap()),
                         regex_search_limit: pool_config.regex_search_limit.unwrap_or(1000),
+                        default_shard: pool_config.default_shard,
                         auth_query: pool_config.auth_query.clone(),
                         auth_query_user: pool_config.auth_query_user.clone(),
                         auth_query_password: pool_config.auth_query_password.clone(),
@@ -473,17 +558,23 @@ impl ConnectionPool {
                             Some(ref plugins) => Some(plugins.clone()),
                             None => config.plugins.clone(),
                         },
-                    },
+                    }),
                     validated: Arc::new(AtomicBool::new(false)),
                     paused: Arc::new(AtomicBool::new(false)),
                     paused_waiter: Arc::new(Notify::new()),
+                    prepared_statement_cache: match pool_config.prepared_statements_cache_size {
+                        0 => None,
+                        _ => Some(Arc::new(Mutex::new(PreparedStatementCache::new(
+                            pool_config.prepared_statements_cache_size,
+                        )))),
+                    },
                 };
 
                 // Connect to the servers to make sure pool configuration is valid
                 // before setting it globally.
                 // Do this async and somewhere else, we don't have to wait here.
                 if config.general.validate_config {
-                    let mut validate_pool = pool.clone();
+                    let validate_pool = pool.clone();
                     tokio::task::spawn(async move {
                         let _ = validate_pool.validate().await;
                     });
@@ -504,7 +595,7 @@ impl ConnectionPool {
     /// when they connect.
     /// This also warms up the pool for clients that connect when
     /// the pooler starts up.
-    pub async fn validate(&mut self) -> Result<(), Error> {
+    pub async fn validate(&self) -> Result<(), Error> {
         let mut futures = Vec::new();
         let validated = Arc::clone(&self.validated);
 
@@ -512,7 +603,7 @@ impl ConnectionPool {
             for server in 0..self.servers(shard) {
                 let databases = self.databases.clone();
                 let validated = Arc::clone(&validated);
-                let pool_server_info = Arc::clone(&self.server_info);
+                let pool_server_parameters = Arc::clone(&self.original_server_parameters);
 
                 let task = tokio::task::spawn(async move {
                     let connection = match databases[shard][server].get().await {
@@ -525,11 +616,10 @@ impl ConnectionPool {
 
                     let proxy = connection;
                     let server = &*proxy;
-                    let server_info = server.server_info();
+                    let server_parameters: ServerParameters = server.server_parameters();
 
-                    let mut guard = pool_server_info.write();
-                    guard.clear();
-                    guard.put(server_info.clone());
+                    let mut guard = pool_server_parameters.write();
+                    *guard = server_parameters;
                     validated.store(true, Ordering::Relaxed);
                 });
 
@@ -541,7 +631,7 @@ impl ConnectionPool {
 
         // TODO: compare server information to make sure
         // all shards are running identical configurations.
-        if self.server_info.read().is_empty() {
+        if !self.validated() {
             error!("Could not validate connection pool");
             return Err(Error::AllServersDown);
         }
@@ -588,19 +678,51 @@ impl ConnectionPool {
     /// Get a connection from the pool.
     pub async fn get(
         &self,
-        shard: usize,               // shard number
+        shard: Option<usize>,       // shard number
         role: Option<Role>,         // primary or replica
         client_stats: &ClientStats, // client id
     ) -> Result<(PooledConnection<'_, ServerPool>, Address), Error> {
-        let mut candidates: Vec<&Address> = self.addresses[shard]
-            .iter()
-            .filter(|address| address.role == role)
-            .collect();
+        let effective_shard_id = if self.shards() == 1 {
+            // The base, unsharded case
+            Some(0)
+        } else {
+            if !self.valid_shard_id(shard) {
+                // None is valid shard ID so it is safe to unwrap here
+                return Err(Error::InvalidShardId(shard.unwrap()));
+            }
+            shard
+        };
 
-        // We shuffle even if least_outstanding_queries is used to avoid imbalance
-        // in cases where all candidates have more or less the same number of outstanding
-        // queries
+        let mut candidates = self
+            .addresses
+            .iter()
+            .flatten()
+            .filter(|address| address.role == role)
+            .collect::<Vec<&Address>>();
+
+        // We start with a shuffled list of addresses even if we end up resorting
+        // this is meant to avoid hitting instance 0 everytime if the sorting metric
+        // ends up being the same for all instances
         candidates.shuffle(&mut thread_rng());
+
+        match effective_shard_id {
+            Some(shard_id) => candidates.retain(|address| address.shard == shard_id),
+            None => match self.settings.default_shard {
+                DefaultShard::Shard(shard_id) => {
+                    candidates.retain(|address| address.shard == shard_id)
+                }
+                DefaultShard::Random => (),
+                DefaultShard::RandomHealthy => {
+                    candidates.sort_by(|a, b| {
+                        b.error_count
+                            .load(Ordering::Relaxed)
+                            .partial_cmp(&a.error_count.load(Ordering::Relaxed))
+                            .unwrap()
+                    });
+                }
+            },
+        };
+
         if self.settings.load_balancing_mode == LoadBalancingMode::LeastOutstandingConnections {
             candidates.sort_by(|a, b| {
                 self.busy_connection_count(b)
@@ -609,6 +731,10 @@ impl ConnectionPool {
             });
         }
 
+        // Indicate we're waiting on a server connection from a pool.
+        let now = Instant::now();
+        client_stats.waiting();
+
         while !candidates.is_empty() {
             // Get the next candidate
             let address = match candidates.pop() {
@@ -619,7 +745,7 @@ impl ConnectionPool {
             let mut force_healthcheck = false;
 
             if self.is_banned(address) {
-                if self.try_unban(&address).await {
+                if self.try_unban(address).await {
                     force_healthcheck = true;
                 } else {
                     debug!("Address {:?} is banned", address);
@@ -627,16 +753,15 @@ impl ConnectionPool {
                 }
             }
 
-            // Indicate we're waiting on a server connection from a pool.
-            let now = Instant::now();
-            client_stats.waiting();
-
             // Check if we can connect
             let mut conn = match self.databases[address.shard][address.address_index]
                 .get()
                 .await
             {
-                Ok(conn) => conn,
+                Ok(conn) => {
+                    address.reset_error_count();
+                    conn
+                }
                 Err(err) => {
                     error!(
                         "Connection checkout error for instance {:?}, error: {:?}",
@@ -662,13 +787,13 @@ impl ConnectionPool {
             // since we last checked the server is ok.
             // Health checks are pretty expensive.
             if !require_healthcheck {
-                let checkout_time: u64 = now.elapsed().as_micros() as u64;
+                let checkout_time = now.elapsed().as_micros() as u64;
                 client_stats.checkout_time(checkout_time);
                 server
                     .stats()
                     .checkout_time(checkout_time, client_stats.application_name());
                 server.stats().active(client_stats.application_name());
-
+                client_stats.active();
                 return Ok((conn, address.clone()));
             }
 
@@ -676,11 +801,24 @@ impl ConnectionPool {
                 .run_health_check(address, server, now, client_stats)
                 .await
             {
+                let checkout_time = now.elapsed().as_micros() as u64;
+                client_stats.checkout_time(checkout_time);
+                server
+                    .stats()
+                    .checkout_time(checkout_time, client_stats.application_name());
+                server.stats().active(client_stats.application_name());
+                client_stats.active();
                 return Ok((conn, address.clone()));
             } else {
                 continue;
             }
         }
+
+        client_stats.idle();
+
+        let checkout_time = now.elapsed().as_micros() as u64;
+        client_stats.checkout_time(checkout_time);
+
         Err(Error::AllServersDown)
     }
 
@@ -735,14 +873,26 @@ impl ConnectionPool {
         // Don't leave a bad connection in the pool.
         server.mark_bad();
 
-        self.ban(&address, BanReason::FailedHealthCheck, Some(client_info));
-        return false;
+        self.ban(address, BanReason::FailedHealthCheck, Some(client_info));
+        false
     }
 
     /// Ban an address (i.e. replica). It no longer will serve
     /// traffic for any new transactions. Existing transactions on that replica
     /// will finish successfully or error out to the clients.
     pub fn ban(&self, address: &Address, reason: BanReason, client_info: Option<&ClientStats>) {
+        // Count the number of errors since the last successful checkout
+        // This is used to determine if the shard is down
+        match reason {
+            BanReason::FailedHealthCheck
+            | BanReason::FailedCheckout
+            | BanReason::MessageSendFailed
+            | BanReason::MessageReceiveFailed => {
+                address.increment_error_count();
+            }
+            _ => (),
+        };
+
         // Primary can never be banned
         if address.role == Role::Primary {
             return;
@@ -848,10 +998,10 @@ impl ConnectionPool {
         let guard = self.banlist.read();
         for banlist in guard.iter() {
             for (address, (reason, timestamp)) in banlist.iter() {
-                bans.push((address.clone(), (reason.clone(), timestamp.clone())));
+                bans.push((address.clone(), (reason.clone(), *timestamp)));
             }
         }
-        return bans;
+        bans
     }
 
     /// Get the address from the host url
@@ -893,10 +1043,11 @@ impl ConnectionPool {
         &self.addresses[shard][server]
     }
 
-    pub fn server_info(&self) -> BytesMut {
-        self.server_info.read().clone()
+    pub fn server_parameters(&self) -> ServerParameters {
+        self.original_server_parameters.read().clone()
     }
 
+    /// Get the number of checked out connection for an address
     fn busy_connection_count(&self, address: &Address) -> u32 {
         let state = self.pool_state(address.shard, address.address_index);
         let idle = state.idle_connections;
@@ -908,39 +1059,93 @@ impl ConnectionPool {
         }
         let busy = provisioned - idle;
         debug!("{:?} has {:?} busy connections", address, busy);
-        return busy;
+        busy
+    }
+
+    fn valid_shard_id(&self, shard: Option<usize>) -> bool {
+        match shard {
+            None => true,
+            Some(shard) => shard < self.shards(),
+        }
+    }
+
+    /// Register a parse statement to the pool's cache and return the rewritten parse
+    ///
+    /// Do not pass an anonymous parse statement to this function
+    pub fn register_parse_to_cache(&self, hash: u64, parse: &Parse) -> Option<Arc<Parse>> {
+        // We should only be calling this function if the cache is enabled
+        match self.prepared_statement_cache {
+            Some(ref prepared_statement_cache) => {
+                let mut cache = prepared_statement_cache.lock();
+                Some(cache.get_or_insert(parse, hash))
+            }
+            None => None,
+        }
+    }
+
+    /// Promote a prepared statement hash in the LRU
+    pub fn promote_prepared_statement_hash(&self, hash: &u64) {
+        // We should only be calling this function if the cache is enabled
+        if let Some(ref prepared_statement_cache) = self.prepared_statement_cache {
+            let mut cache = prepared_statement_cache.lock();
+            cache.promote(hash);
+        }
     }
 }
 
 /// Wrapper for the bb8 connection pool.
 pub struct ServerPool {
+    /// Server address.
     address: Address,
+
+    /// Server Postgres user.
     user: User,
+
+    /// Server database.
     database: String,
+
+    /// Client/server mapping.
     client_server_map: ClientServerMap,
-    stats: Arc<PoolStats>,
+
+    /// Server auth hash (for auth passthrough).
     auth_hash: Arc<RwLock<Option<String>>>,
+
+    /// Server plugins.
     plugins: Option<Plugins>,
+
+    /// Should we clean up dirty connections before putting them into the pool?
+    cleanup_connections: bool,
+
+    /// Log client parameter status changes
+    log_client_parameter_status_changes: bool,
+
+    /// Prepared statement cache size
+    prepared_statement_cache_size: usize,
 }
 
 impl ServerPool {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         address: Address,
         user: User,
         database: &str,
         client_server_map: ClientServerMap,
-        stats: Arc<PoolStats>,
         auth_hash: Arc<RwLock<Option<String>>>,
         plugins: Option<Plugins>,
+        cleanup_connections: bool,
+        log_client_parameter_status_changes: bool,
+        prepared_statement_cache_size: usize,
     ) -> ServerPool {
         ServerPool {
             address,
-            user: user.clone(),
+            user,
             database: database.to_string(),
             client_server_map,
-            stats,
             auth_hash,
             plugins,
+            cleanup_connections,
+            log_client_parameter_status_changes,
+            prepared_statement_cache_size,
         }
     }
 }
@@ -956,7 +1161,6 @@ impl ManageConnection for ServerPool {
 
         let stats = Arc::new(ServerStats::new(
             self.address.clone(),
-            self.stats.clone(),
             tokio::time::Instant::now(),
         ));
 
@@ -970,6 +1174,9 @@ impl ManageConnection for ServerPool {
             self.client_server_map.clone(),
             stats.clone(),
             self.auth_hash.clone(),
+            self.cleanup_connections,
+            self.log_client_parameter_status_changes,
+            self.prepared_statement_cache_size,
         )
         .await
         {

src/prometheus.rs

@@ -1,6 +1,6 @@
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
-use log::{error, info, warn};
+use log::{debug, error, info};
 use phf::phf_map;
 use std::collections::HashMap;
 use std::fmt;
@@ -9,8 +9,9 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use crate::config::Address;
-use crate::pool::get_all_pools;
-use crate::stats::{get_pool_stats, get_server_stats, ServerStats};
+use crate::pool::{get_all_pools, PoolIdentifier};
+use crate::stats::pool::PoolStats;
+use crate::stats::{get_server_stats, ServerStats};
 
 struct MetricHelpType {
     help: &'static str,
@@ -233,10 +234,10 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         Self::from_name(&format!("stats_{}", name), value, labels)
     }
 
-    fn from_pool(pool: &(String, String), name: &str, value: u64) -> Option<PrometheusMetric<u64>> {
+    fn from_pool(pool_id: PoolIdentifier, name: &str, value: u64) -> Option<PrometheusMetric<u64>> {
         let mut labels = HashMap::new();
-        labels.insert("pool", pool.0.clone());
-        labels.insert("user", pool.1.clone());
+        labels.insert("pool", pool_id.db);
+        labels.insert("user", pool_id.user);
 
         Self::from_name(&format!("pools_{}", name), value, labels)
     }
@@ -274,7 +275,7 @@ fn push_address_stats(lines: &mut Vec<String>) {
                     {
                         lines.push(prometheus_metric.to_string());
                     } else {
-                        warn!("Metric {} not implemented for {}", key, address.name());
+                        debug!("Metric {} not implemented for {}", key, address.name());
                     }
                 }
             }
@@ -284,18 +285,15 @@ fn push_address_stats(lines: &mut Vec<String>) {
 
 // Adds relevant metrics shown in a SHOW POOLS admin command.
 fn push_pool_stats(lines: &mut Vec<String>) {
-    let pool_stats = get_pool_stats();
-    for (pool, stats) in pool_stats.iter() {
-        let stats = &**stats;
+    let pool_stats = PoolStats::construct_pool_lookup();
+    for (pool_id, stats) in pool_stats.iter() {
         for (name, value) in stats.clone() {
-            if let Some(prometheus_metric) = PrometheusMetric::<u64>::from_pool(pool, &name, value)
+            if let Some(prometheus_metric) =
+                PrometheusMetric::<u64>::from_pool(pool_id.clone(), &name, value)
             {
                 lines.push(prometheus_metric.to_string());
             } else {
-                warn!(
-                    "Metric {} not implemented for ({},{})",
-                    name, pool.0, pool.1
-                );
+                debug!("Metric {} not implemented for ({})", name, *pool_id);
             }
         }
     }
@@ -320,7 +318,7 @@ fn push_database_stats(lines: &mut Vec<String>) {
                     {
                         lines.push(prometheus_metric.to_string());
                     } else {
-                        warn!("Metric {} not implemented for {}", key, address.name());
+                        debug!("Metric {} not implemented for {}", key, address.name());
                     }
                 }
             }
@@ -366,7 +364,7 @@ fn push_server_stats(lines: &mut Vec<String>) {
                         {
                             lines.push(prometheus_metric.to_string());
                         } else {
-                            warn!("Metric {} not implemented for {}", key, address.name());
+                            debug!("Metric {} not implemented for {}", key, address.name());
                         }
                     }
                 }

src/scram.rs

@@ -79,12 +79,12 @@ impl ScramSha256 {
         let server_message = Message::parse(message)?;
 
         if !server_message.nonce.starts_with(&self.nonce) {
-            return Err(Error::ProtocolSyncError(format!("SCRAM")));
+            return Err(Error::ProtocolSyncError("SCRAM".to_string()));
         }
 
         let salt = match general_purpose::STANDARD.decode(&server_message.salt) {
             Ok(salt) => salt,
-            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
+            Err(_) => return Err(Error::ProtocolSyncError("SCRAM".to_string())),
         };
 
         let salted_password = Self::hi(
@@ -166,9 +166,9 @@ impl ScramSha256 {
     pub fn finish(&mut self, message: &BytesMut) -> Result<(), Error> {
         let final_message = FinalMessage::parse(message)?;
 
-        let verifier = match general_purpose::STANDARD.decode(&final_message.value) {
+        let verifier = match general_purpose::STANDARD.decode(final_message.value) {
             Ok(verifier) => verifier,
-            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
+            Err(_) => return Err(Error::ProtocolSyncError("SCRAM".to_string())),
         };
 
         let mut hmac = match Hmac::<Sha256>::new_from_slice(&self.salted_password) {
@@ -230,14 +230,14 @@ impl Message {
             .collect::<Vec<String>>();
 
         if parts.len() != 3 {
-            return Err(Error::ProtocolSyncError(format!("SCRAM")));
+            return Err(Error::ProtocolSyncError("SCRAM".to_string()));
         }
 
         let nonce = str::replace(&parts[0], "r=", "");
         let salt = str::replace(&parts[1], "s=", "");
         let iterations = match str::replace(&parts[2], "i=", "").parse::<u32>() {
             Ok(iterations) => iterations,
-            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
+            Err(_) => return Err(Error::ProtocolSyncError("SCRAM".to_string())),
         };
 
         Ok(Message {
@@ -257,7 +257,7 @@ impl FinalMessage {
     /// Parse the server final validation message.
     pub fn parse(message: &BytesMut) -> Result<FinalMessage, Error> {
         if !message.starts_with(b"v=") || message.len() < 4 {
-            return Err(Error::ProtocolSyncError(format!("SCRAM")));
+            return Err(Error::ProtocolSyncError("SCRAM".to_string()));
         }
 
         Ok(FinalMessage {

src/server.rs

@@ -3,11 +3,14 @@
 use bytes::{Buf, BufMut, BytesMut};
 use fallible_iterator::FallibleIterator;
 use log::{debug, error, info, trace, warn};
+use lru::LruCache;
+use once_cell::sync::Lazy;
 use parking_lot::{Mutex, RwLock};
 use postgres_protocol::message;
-use std::collections::HashMap;
-use std::io::Read;
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::mem;
 use std::net::IpAddr;
+use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::SystemTime;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, BufStream};
@@ -19,6 +22,7 @@ use crate::config::{get_config, Address, User};
 use crate::constants::*;
 use crate::dns_cache::{AddrSet, CACHED_RESOLVER};
 use crate::errors::{Error, ServerIdentifier};
+use crate::messages::BytesMutReader;
 use crate::messages::*;
 use crate::mirrors::MirroringManager;
 use crate::pool::ClientServerMap;
@@ -105,10 +109,10 @@ impl StreamInner {
 
 #[derive(Copy, Clone)]
 struct CleanupState {
-    /// If server connection requires DISCARD ALL before checkin because of set statement
+    /// If server connection requires RESET ALL before checkin because of set statement
     needs_cleanup_set: bool,
 
-    /// If server connection requires DISCARD ALL before checkin because of prepare statement
+    /// If server connection requires DEALLOCATE ALL before checkin because of prepare statement
     needs_cleanup_prepare: bool,
 }
 
@@ -145,6 +149,120 @@ impl std::fmt::Display for CleanupState {
     }
 }
 
+static TRACKED_PARAMETERS: Lazy<HashSet<String>> = Lazy::new(|| {
+    let mut set = HashSet::new();
+    set.insert("client_encoding".to_string());
+    set.insert("DateStyle".to_string());
+    set.insert("TimeZone".to_string());
+    set.insert("standard_conforming_strings".to_string());
+    set.insert("application_name".to_string());
+    set
+});
+
+#[derive(Debug, Clone)]
+pub struct ServerParameters {
+    parameters: HashMap<String, String>,
+}
+
+impl Default for ServerParameters {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ServerParameters {
+    pub fn new() -> Self {
+        let mut server_parameters = ServerParameters {
+            parameters: HashMap::new(),
+        };
+
+        server_parameters.set_param("client_encoding".to_string(), "UTF8".to_string(), false);
+        server_parameters.set_param("DateStyle".to_string(), "ISO, MDY".to_string(), false);
+        server_parameters.set_param("TimeZone".to_string(), "Etc/UTC".to_string(), false);
+        server_parameters.set_param(
+            "standard_conforming_strings".to_string(),
+            "on".to_string(),
+            false,
+        );
+        server_parameters.set_param("application_name".to_string(), "pgcat".to_string(), false);
+
+        server_parameters
+    }
+
+    /// returns true if a tracked parameter was set, false if it was a non-tracked parameter
+    /// if startup is false, then then only tracked parameters will be set
+    pub fn set_param(&mut self, mut key: String, value: String, startup: bool) {
+        // The startup parameter will send uncapitalized keys but parameter status packets will send capitalized keys
+        if key == "timezone" {
+            key = "TimeZone".to_string();
+        } else if key == "datestyle" {
+            key = "DateStyle".to_string();
+        };
+
+        if TRACKED_PARAMETERS.contains(&key) || startup {
+            self.parameters.insert(key, value);
+        }
+    }
+
+    pub fn set_from_hashmap(&mut self, parameters: &HashMap<String, String>, startup: bool) {
+        // iterate through each and call set_param
+        for (key, value) in parameters {
+            self.set_param(key.to_string(), value.to_string(), startup);
+        }
+    }
+
+    // Gets the diff of the parameters
+    fn compare_params(&self, incoming_parameters: &ServerParameters) -> HashMap<String, String> {
+        let mut diff = HashMap::new();
+
+        // iterate through tracked parameters
+        for key in TRACKED_PARAMETERS.iter() {
+            if let Some(incoming_value) = incoming_parameters.parameters.get(key) {
+                if let Some(value) = self.parameters.get(key) {
+                    if value != incoming_value {
+                        diff.insert(key.to_string(), incoming_value.to_string());
+                    }
+                }
+            }
+        }
+
+        diff
+    }
+
+    pub fn get_application_name(&self) -> &String {
+        // Can unwrap because we set it in the constructor
+        self.parameters.get("application_name").unwrap()
+    }
+
+    fn add_parameter_message(key: &str, value: &str, buffer: &mut BytesMut) {
+        buffer.put_u8(b'S');
+
+        // 4 is len of i32, the plus for the null terminator
+        let len = 4 + key.len() + 1 + value.len() + 1;
+
+        buffer.put_i32(len as i32);
+
+        buffer.put_slice(key.as_bytes());
+        buffer.put_u8(0);
+        buffer.put_slice(value.as_bytes());
+        buffer.put_u8(0);
+    }
+}
+
+impl From<&ServerParameters> for BytesMut {
+    fn from(server_parameters: &ServerParameters) -> Self {
+        let mut bytes = BytesMut::new();
+
+        for (key, value) in &server_parameters.parameters {
+            ServerParameters::add_parameter_message(key, value, &mut bytes);
+        }
+
+        bytes
+    }
+}
+
+// pub fn compare
+
 /// Server state.
 pub struct Server {
     /// Server host, e.g. localhost,
@@ -158,7 +276,7 @@ pub struct Server {
     buffer: BytesMut,
 
     /// Server information the server sent us over on startup.
-    server_info: BytesMut,
+    server_parameters: ServerParameters,
 
     /// Backend id and secret key used for query cancellation.
     process_id: i32,
@@ -170,10 +288,13 @@ pub struct Server {
     /// Is there more data for the client to read.
     data_available: bool,
 
+    /// Is the server in copy-in or copy-out modes
+    in_copy_mode: bool,
+
     /// Is the server broken? We'll remote it from the pool if so.
     bad: bool,
 
-    /// If server connection requires DISCARD ALL before checkin
+    /// If server connection requires reset statements before checkin
     cleanup_state: CleanupState,
 
     /// Mapping of clients and servers used for query cancellation.
@@ -188,18 +309,31 @@ pub struct Server {
     /// Application name using the server at the moment.
     application_name: String,
 
-    // Last time that a successful server send or response happened
+    /// Last time that a successful server send or response happened
     last_activity: SystemTime,
 
     mirror_manager: Option<MirroringManager>,
 
-    // Associated addresses used
+    /// Associated addresses used
     addr_set: Option<AddrSet>,
+
+    /// Should clean up dirty connections?
+    cleanup_connections: bool,
+
+    /// Log client parameter status changes
+    log_client_parameter_status_changes: bool,
+
+    /// Prepared statements
+    prepared_statement_cache: Option<LruCache<String, ()>>,
+
+    /// Prepared statement being currently registered on the server.
+    registering_prepared_statement: VecDeque<String>,
 }
 
 impl Server {
     /// Pretend to be the Postgres client and connect to the server given host, port and credentials.
     /// Perform the authentication and return the server in a ready for query state.
+    #[allow(clippy::too_many_arguments)]
     pub async fn startup(
         address: &Address,
         user: &User,
@@ -207,6 +341,9 @@ impl Server {
         client_server_map: ClientServerMap,
         stats: Arc<ServerStats>,
         auth_hash: Arc<RwLock<Option<String>>>,
+        cleanup_connections: bool,
+        log_client_parameter_status_changes: bool,
+        prepared_statement_cache_size: usize,
     ) -> Result<Server, Error> {
         let cached_resolver = CACHED_RESOLVER.load();
         let mut addr_set: Option<AddrSet> = None;
@@ -306,10 +443,7 @@ impl Server {
 
                 // Something else?
                 m => {
-                    return Err(Error::SocketError(format!(
-                        "Unknown message: {}",
-                        m as char
-                    )));
+                    return Err(Error::SocketError(format!("Unknown message: {}", { m })));
                 }
             }
         } else {
@@ -327,27 +461,22 @@ impl Server {
             None => &user.username,
         };
 
-        let password = match user.server_password {
-            Some(ref server_password) => Some(server_password),
-            None => match user.password {
-                Some(ref password) => Some(password),
-                None => None,
-            },
+        let password = match user.server_password.as_ref() {
+            Some(server_password) => Some(server_password),
+            None => user.password.as_ref(),
         };
 
         startup(&mut stream, username, database).await?;
 
-        let mut server_info = BytesMut::new();
         let mut process_id: i32 = 0;
         let mut secret_key: i32 = 0;
-        let server_identifier = ServerIdentifier::new(username, &database);
+        let server_identifier = ServerIdentifier::new(username, database);
 
         // We'll be handling multiple packets, but they will all be structured the same.
         // We'll loop here until this exchange is complete.
-        let mut scram: Option<ScramSha256> = match password {
-            Some(password) => Some(ScramSha256::new(password)),
-            None => None,
-        };
+        let mut scram: Option<ScramSha256> = password.map(|password| ScramSha256::new(password));
+
+        let mut server_parameters = ServerParameters::new();
 
         loop {
             let code = match stream.read_u8().await {
@@ -578,8 +707,7 @@ impl Server {
 
                         // An error message will be present.
                         _ => {
-                            // Read the error message without the terminating null character.
-                            let mut error = vec![0u8; len as usize - 4 - 1];
+                            let mut error = vec![0u8; len as usize];
 
                             match stream.read_exact(&mut error).await {
                                 Ok(_) => (),
@@ -591,10 +719,14 @@ impl Server {
                                 }
                             };
 
-                            // TODO: the error message contains multiple fields; we can decode them and
-                            // present a prettier message to the user.
-                            // See: https://www.postgresql.org/docs/12/protocol-error-fields.html
-                            error!("Server error: {}", String::from_utf8_lossy(&error));
+                            let fields = match PgErrorMsg::parse(&error) {
+                                Ok(f) => f,
+                                Err(err) => {
+                                    return Err(err);
+                                }
+                            };
+                            trace!("error fields: {}", &fields);
+                            error!("server error: {}: {}", fields.severity, fields.message);
                         }
                     };
 
@@ -603,9 +735,10 @@ impl Server {
 
                 // ParameterStatus
                 'S' => {
-                    let mut param = vec![0u8; len as usize - 4];
+                    let mut bytes = BytesMut::with_capacity(len as usize - 4);
+                    bytes.resize(len as usize - mem::size_of::<i32>(), b'0');
 
-                    match stream.read_exact(&mut param).await {
+                    match stream.read_exact(&mut bytes[..]).await {
                         Ok(_) => (),
                         Err(_) => {
                             return Err(Error::ServerStartupError(
@@ -615,12 +748,13 @@ impl Server {
                         }
                     };
 
+                    let key = bytes.read_string().unwrap();
+                    let value = bytes.read_string().unwrap();
+
                     // Save the parameter so we can pass it to the client later.
                     // These can be server_encoding, client_encoding, server timezone, Postgres version,
                     // and many more interesting things we should know about the Postgres server we are talking to.
-                    server_info.put_u8(b'S');
-                    server_info.put_i32(len);
-                    server_info.put_slice(&param[..]);
+                    server_parameters.set_param(key, value, true);
                 }
 
                 // BackendKeyData
@@ -662,14 +796,15 @@ impl Server {
                         }
                     };
 
-                    let mut server = Server {
+                    let server = Server {
                         address: address.clone(),
                         stream: BufStream::new(stream),
                         buffer: BytesMut::with_capacity(8196),
-                        server_info,
+                        server_parameters,
                         process_id,
                         secret_key,
                         in_transaction: false,
+                        in_copy_mode: false,
                         data_available: false,
                         bad: false,
                         cleanup_state: CleanupState::new(),
@@ -677,7 +812,7 @@ impl Server {
                         addr_set,
                         connected_at: chrono::offset::Utc::now().naive_utc(),
                         stats,
-                        application_name: String::new(),
+                        application_name: "pgcat".to_string(),
                         last_activity: SystemTime::now(),
                         mirror_manager: match address.mirrors.len() {
                             0 => None,
@@ -687,10 +822,17 @@ impl Server {
                                 address.mirrors.clone(),
                             )),
                         },
+                        cleanup_connections,
+                        log_client_parameter_status_changes,
+                        prepared_statement_cache: match prepared_statement_cache_size {
+                            0 => None,
+                            _ => Some(LruCache::new(
+                                NonZeroUsize::new(prepared_statement_cache_size).unwrap(),
+                            )),
+                        },
+                        registering_prepared_statement: VecDeque::new(),
                     };
 
-                    server.set_name("pgcat").await?;
-
                     return Ok(server);
                 }
 
@@ -740,7 +882,7 @@ impl Server {
         self.mirror_send(messages);
         self.stats().data_sent(messages.len());
 
-        match write_all_flush(&mut self.stream, &messages).await {
+        match write_all_flush(&mut self.stream, messages).await {
             Ok(_) => {
                 // Successfully sent to server
                 self.last_activity = SystemTime::now();
@@ -760,7 +902,10 @@ impl Server {
     /// Receive data from the server in response to a client request.
     /// This method must be called multiple times while `self.is_data_available()` is true
     /// in order to receive all data the server has to offer.
-    pub async fn recv(&mut self) -> Result<BytesMut, Error> {
+    pub async fn recv(
+        &mut self,
+        mut client_server_parameters: Option<&mut ServerParameters>,
+    ) -> Result<BytesMut, Error> {
         loop {
             let mut message = match read_message(&mut self.stream).await {
                 Ok(message) => message,
@@ -815,30 +960,71 @@ impl Server {
 
                     // There is no more data available from the server.
                     self.data_available = false;
-
                     break;
                 }
 
+                // ErrorResponse
+                'E' => {
+                    if self.in_copy_mode {
+                        self.in_copy_mode = false;
+                    }
+
+                    // Remove the prepared statement from the cache, it has a syntax error or something else bad happened.
+                    if let Some(prepared_stmt_name) =
+                        self.registering_prepared_statement.pop_front()
+                    {
+                        if let Some(ref mut cache) = self.prepared_statement_cache {
+                            if let Some(_removed) = cache.pop(&prepared_stmt_name) {
+                                debug!(
+                                    "Removed {} from prepared statement cache",
+                                    prepared_stmt_name
+                                );
+                            } else {
+                                // Shouldn't happen.
+                                debug!("Prepared statement {} was not cached", prepared_stmt_name);
+                            }
+                        }
+                    }
+
+                    if self.prepared_statement_cache.is_some() {
+                        let error_message = PgErrorMsg::parse(&message)?;
+                        if error_message.message == "cached plan must not change result type" {
+                            warn!("Server {:?} changed schema, dropping connection to clean up prepared statements", self.address);
+                            // This will still result in an error to the client, but this server connection will drop all cached prepared statements
+                            // so that any new queries will be re-prepared
+                            // TODO: Other ideas to solve errors when there are DDL changes after a statement has been prepared
+                            //  - Recreate entire connection pool to force recreation of all server connections
+                            //  - Clear the ConnectionPool's statement cache so that new statement names are generated
+                            //  - Implement a retry (re-prepare) so the client doesn't see an error
+                            self.cleanup_state.needs_cleanup_prepare = true;
+                        }
+                    }
+                }
+
                 // CommandComplete
                 'C' => {
-                    let mut command_tag = String::new();
-                    match message.reader().read_to_string(&mut command_tag) {
-                        Ok(_) => {
+                    if self.in_copy_mode {
+                        self.in_copy_mode = false;
+                    }
+
+                    match message.read_string() {
+                        Ok(command) => {
                             // Non-exhaustive list of commands that are likely to change session variables/resources
                             // which can leak between clients. This is a best effort to block bad clients
                             // from poisoning a transaction-mode pool by setting inappropriate session variables
-                            match command_tag.as_str() {
-                                "SET\0" => {
+                            match command.as_str() {
+                                "SET" => {
                                     // We don't detect set statements in transactions
                                     // No great way to differentiate between set and set local
                                     // As a result, we will miss cases when set statements are used in transactions
-                                    // This will reduce amount of discard statements sent
+                                    // This will reduce amount of reset statements sent
                                     if !self.in_transaction {
                                         debug!("Server connection marked for clean up");
                                         self.cleanup_state.needs_cleanup_set = true;
                                     }
                                 }
-                                "PREPARE\0" => {
+
+                                "PREPARE" => {
                                     debug!("Server connection marked for clean up");
                                     self.cleanup_state.needs_cleanup_prepare = true;
                                 }
@@ -852,6 +1038,20 @@ impl Server {
                     }
                 }
 
+                'S' => {
+                    let key = message.read_string().unwrap();
+                    let value = message.read_string().unwrap();
+
+                    if let Some(client_server_parameters) = client_server_parameters.as_mut() {
+                        client_server_parameters.set_param(key.clone(), value.clone(), false);
+                        if self.log_client_parameter_status_changes {
+                            info!("Client parameter status change: {} = {}", key, value)
+                        }
+                    }
+
+                    self.server_parameters.set_param(key, value, false);
+                }
+
                 // DataRow
                 'D' => {
                     // More data is available after this message, this is not the end of the reply.
@@ -864,10 +1064,14 @@ impl Server {
                 }
 
                 // CopyInResponse: copy is starting from client to server.
-                'G' => break,
+                'G' => {
+                    self.in_copy_mode = true;
+                    break;
+                }
 
                 // CopyOutResponse: copy is starting from the server to the client.
                 'H' => {
+                    self.in_copy_mode = true;
                     self.data_available = true;
                     break;
                 }
@@ -884,6 +1088,11 @@ impl Server {
                 // Buffer until ReadyForQuery shows up, so don't exit the loop yet.
                 'c' => (),
 
+                // Parse complete successfully
+                '1' => {
+                    self.registering_prepared_statement.pop_front();
+                }
+
                 // Anything else, e.g. errors, notices, etc.
                 // Keep buffering until ReadyForQuery shows up.
                 _ => (),
@@ -905,6 +1114,105 @@ impl Server {
         Ok(bytes)
     }
 
+    // Determines if the server already has a prepared statement with the given name
+    // Increments the prepared statement cache hit counter
+    pub fn has_prepared_statement(&mut self, name: &str) -> bool {
+        let cache = match &mut self.prepared_statement_cache {
+            Some(cache) => cache,
+            None => return false,
+        };
+
+        let has_it = cache.get(name).is_some();
+        if has_it {
+            self.stats.prepared_cache_hit();
+        } else {
+            self.stats.prepared_cache_miss();
+        }
+
+        has_it
+    }
+
+    fn add_prepared_statement_to_cache(&mut self, name: &str) -> Option<String> {
+        let cache = match &mut self.prepared_statement_cache {
+            Some(cache) => cache,
+            None => return None,
+        };
+
+        self.stats.prepared_cache_add();
+
+        // If we evict something, we need to close it on the server
+        if let Some((evicted_name, _)) = cache.push(name.to_string(), ()) {
+            if evicted_name != name {
+                debug!(
+                    "Evicted prepared statement {} from cache, replaced with {}",
+                    evicted_name, name
+                );
+                return Some(evicted_name);
+            }
+        };
+
+        None
+    }
+
+    fn remove_prepared_statement_from_cache(&mut self, name: &str) {
+        let cache = match &mut self.prepared_statement_cache {
+            Some(cache) => cache,
+            None => return,
+        };
+
+        self.stats.prepared_cache_remove();
+        cache.pop(name);
+    }
+
+    pub async fn register_prepared_statement(
+        &mut self,
+        parse: &Parse,
+        should_send_parse_to_server: bool,
+    ) -> Result<(), Error> {
+        if !self.has_prepared_statement(&parse.name) {
+            self.registering_prepared_statement
+                .push_back(parse.name.clone());
+
+            let mut bytes = BytesMut::new();
+
+            if should_send_parse_to_server {
+                let parse_bytes: BytesMut = parse.try_into()?;
+                bytes.extend_from_slice(&parse_bytes);
+            }
+
+            // If we evict something, we need to close it on the server
+            // We do this by adding it to the messages we're sending to the server before the sync
+            if let Some(evicted_name) = self.add_prepared_statement_to_cache(&parse.name) {
+                self.remove_prepared_statement_from_cache(&evicted_name);
+                let close_bytes: BytesMut = Close::new(&evicted_name).try_into()?;
+                bytes.extend_from_slice(&close_bytes);
+            };
+
+            // If we have a parse or close we need to send to the server, send them and sync
+            if !bytes.is_empty() {
+                bytes.extend_from_slice(&sync());
+
+                self.send(&bytes).await?;
+
+                loop {
+                    self.recv(None).await?;
+
+                    if !self.is_data_available() {
+                        break;
+                    }
+                }
+            }
+        };
+
+        // If it's not there, something went bad, I'm guessing bad syntax or permissions error
+        // on the server.
+        if !self.has_prepared_statement(&parse.name) {
+            Err(Error::PreparedStatementError)
+        } else {
+            Ok(())
+        }
+    }
+
     /// If the server is still inside a transaction.
     /// If the client disconnects while the server is in a transaction, we will clean it up.
     pub fn in_transaction(&self) -> bool {
@@ -912,6 +1220,11 @@ impl Server {
         self.in_transaction
     }
 
+    /// Currently copying data from client to server or vice-versa.
+    pub fn in_copy_mode(&self) -> bool {
+        self.in_copy_mode
+    }
+
     /// We don't buffer all of server responses, e.g. COPY OUT produces too much data.
     /// The client is responsible to call `self.recv()` while this method returns true.
     pub fn is_data_available(&self) -> bool {
@@ -941,9 +1254,28 @@ impl Server {
     }
 
     /// Get server startup information to forward it to the client.
-    /// Not used at the moment.
-    pub fn server_info(&self) -> BytesMut {
-        self.server_info.clone()
+    pub fn server_parameters(&self) -> ServerParameters {
+        self.server_parameters.clone()
+    }
+
+    pub async fn sync_parameters(&mut self, parameters: &ServerParameters) -> Result<(), Error> {
+        let parameter_diff = self.server_parameters.compare_params(parameters);
+
+        if parameter_diff.is_empty() {
+            return Ok(());
+        }
+
+        let mut query = String::from("");
+
+        for (key, value) in parameter_diff {
+            query.push_str(&format!("SET {} TO '{}';", key, value));
+        }
+
+        let res = self.query(&query).await;
+
+        self.cleanup_state.reset();
+
+        res
     }
 
     /// Indicate that this server connection cannot be re-used and must be discarded.
@@ -977,7 +1309,7 @@ impl Server {
         self.send(&query).await?;
 
         loop {
-            let _ = self.recv().await?;
+            let _ = self.recv(None).await?;
 
             if !self.data_available {
                 break;
@@ -995,41 +1327,40 @@ impl Server {
         // server connection thrashing if clients repeatedly do this.
         // Instead, we ROLLBACK that transaction before putting the connection back in the pool
         if self.in_transaction() {
-            warn!("Server returned while still in transaction, rolling back transaction");
+            warn!(target: "pgcat::server::cleanup", "Server returned while still in transaction, rolling back transaction");
             self.query("ROLLBACK").await?;
         }
 
         // Client disconnected but it performed session-altering operations such as
         // SET statement_timeout to 1 or create a prepared statement. We clear that
         // to avoid leaking state between clients. For performance reasons we only
-        // send `DISCARD ALL` if we think the session is altered instead of just sending
+        // send `RESET ALL` if we think the session is altered instead of just sending
         // it before each checkin.
-        if self.cleanup_state.needs_cleanup() {
-            warn!("Server returned with session state altered, discarding state ({}) for application {}", self.cleanup_state, self.application_name);
-            self.query("DISCARD ALL").await?;
-            self.query("RESET ROLE").await?;
+        if self.cleanup_state.needs_cleanup() && self.cleanup_connections {
+            info!(target: "pgcat::server::cleanup", "Server returned with session state altered, discarding state ({}) for application {}", self.cleanup_state, self.application_name);
+            let mut reset_string = String::from("RESET ROLE;");
+
+            if self.cleanup_state.needs_cleanup_set {
+                reset_string.push_str("RESET ALL;");
+            };
+
+            if self.cleanup_state.needs_cleanup_prepare {
+                reset_string.push_str("DEALLOCATE ALL;");
+                // Since we deallocated all prepared statements, we need to clear the cache
+                if let Some(cache) = &mut self.prepared_statement_cache {
+                    cache.clear();
+                }
+            };
+
+            self.query(&reset_string).await?;
             self.cleanup_state.reset();
         }
 
-        Ok(())
-    }
-
-    /// A shorthand for `SET application_name = $1`.
-    pub async fn set_name(&mut self, name: &str) -> Result<(), Error> {
-        if self.application_name != name {
-            self.application_name = name.to_string();
-            // We don't want `SET application_name` to mark the server connection
-            // as needing cleanup
-            let needs_cleanup_before = self.cleanup_state;
-
-            let result = Ok(self
-                .query(&format!("SET application_name = '{}'", name))
-                .await?);
-            self.cleanup_state = needs_cleanup_before;
-            result
-        } else {
-            Ok(())
+        if self.in_copy_mode() {
+            warn!(target: "pgcat::server::cleanup", "Server returned while still in copy-mode");
         }
+
+        Ok(())
     }
 
     /// get Server stats
@@ -1048,22 +1379,20 @@ impl Server {
         self.last_activity
     }
 
-    // Marks a connection as needing DISCARD ALL at checkin
+    // Marks a connection as needing cleanup at checkin
     pub fn mark_dirty(&mut self) {
         self.cleanup_state.set_true();
     }
 
     pub fn mirror_send(&mut self, bytes: &BytesMut) {
-        match self.mirror_manager.as_mut() {
-            Some(manager) => manager.send(bytes),
-            None => (),
+        if let Some(manager) = self.mirror_manager.as_mut() {
+            manager.send(bytes)
         }
     }
 
     pub fn mirror_disconnect(&mut self) {
-        match self.mirror_manager.as_mut() {
-            Some(manager) => manager.disconnect(),
-            None => (),
+        if let Some(manager) = self.mirror_manager.as_mut() {
+            manager.disconnect()
         }
     }
 
@@ -1084,13 +1413,16 @@ impl Server {
             client_server_map,
             Arc::new(ServerStats::default()),
             Arc::new(RwLock::new(None)),
+            true,
+            false,
+            0,
         )
         .await?;
         debug!("Connected!, sending query.");
         server.send(&simple_query(query)).await?;
-        let mut message = server.recv().await?;
+        let mut message = server.recv(None).await?;
 
-        Ok(parse_query_message(&mut message).await?)
+        parse_query_message(&mut message).await
     }
 }
 

src/sharding.rs

@@ -64,7 +64,7 @@ impl Sharder {
     fn sha1(&self, key: i64) -> usize {
         let mut hasher = Sha1::new();
 
-        hasher.update(&key.to_string().as_bytes());
+        hasher.update(key.to_string().as_bytes());
 
         let result = hasher.finalize();
 
@@ -202,10 +202,10 @@ mod test {
     #[test]
     fn test_sha1_hash() {
         let sharder = Sharder::new(12, ShardingFunction::Sha1);
-        let ids = vec![
+        let ids = [
             0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
         ];
-        let shards = vec![
+        let shards = [
             4, 7, 8, 3, 6, 0, 0, 10, 3, 11, 1, 7, 4, 4, 11, 2, 5, 0, 8, 3,
         ];
 

src/stats.rs

@@ -1,4 +1,3 @@
-use crate::pool::PoolIdentifier;
 /// Statistics and reporting.
 use arc_swap::ArcSwap;
 
@@ -16,13 +15,11 @@ pub mod pool;
 pub mod server;
 pub use address::AddressStats;
 pub use client::{ClientState, ClientStats};
-pub use pool::PoolStats;
 pub use server::{ServerState, ServerStats};
 
 /// Convenience types for various stats
 type ClientStatesLookup = HashMap<i32, Arc<ClientStats>>;
 type ServerStatesLookup = HashMap<i32, Arc<ServerStats>>;
-type PoolStatsLookup = HashMap<(String, String), Arc<PoolStats>>;
 
 /// Stats for individual client connections
 /// Used in SHOW CLIENTS.
@@ -34,11 +31,6 @@ static CLIENT_STATS: Lazy<Arc<RwLock<ClientStatesLookup>>> =
 static SERVER_STATS: Lazy<Arc<RwLock<ServerStatesLookup>>> =
     Lazy::new(|| Arc::new(RwLock::new(ServerStatesLookup::default())));
 
-/// Aggregate stats for each pool (a pool is identified by database name and username)
-/// Used in SHOW POOLS.
-static POOL_STATS: Lazy<Arc<RwLock<PoolStatsLookup>>> =
-    Lazy::new(|| Arc::new(RwLock::new(PoolStatsLookup::default())));
-
 /// The statistics reporter. An instance is given to each possible source of statistics,
 /// e.g. client stats, server stats, connection pool stats.
 pub static REPORTER: Lazy<ArcSwap<Reporter>> =
@@ -80,13 +72,6 @@ impl Reporter {
     fn server_disconnecting(&self, server_id: i32) {
         SERVER_STATS.write().remove(&server_id);
     }
-
-    /// Register a pool with the stats system.
-    fn pool_register(&self, identifier: PoolIdentifier, stats: Arc<PoolStats>) {
-        POOL_STATS
-            .write()
-            .insert((identifier.db, identifier.user), stats);
-    }
 }
 
 /// The statistics collector which used for calculating averages
@@ -113,6 +98,7 @@ impl Collector {
                 for stats in server_stats.values() {
                     if !stats.check_address_stat_average_is_updated_status() {
                         stats.address_stats().update_averages();
+                        stats.address_stats().reset_current_counts();
                         stats.set_address_stat_average_is_updated_status(true);
                     }
                 }
@@ -138,12 +124,6 @@ pub fn get_server_stats() -> ServerStatesLookup {
     SERVER_STATS.read().clone()
 }
 
-/// Get a snapshot of pool statistics.
-/// by the `Collector`.
-pub fn get_pool_stats() -> PoolStatsLookup {
-    POOL_STATS.read().clone()
-}
-
 /// Get the statistics reporter used to update stats across the pools/clients.
 pub fn get_reporter() -> Reporter {
     (*(*REPORTER.load())).clone()

src/stats/address.rs

@@ -1,35 +1,26 @@
 use std::sync::atomic::*;
 use std::sync::Arc;
 
+#[derive(Debug, Clone, Default)]
+struct AddressStatFields {
+    xact_count: Arc<AtomicU64>,
+    query_count: Arc<AtomicU64>,
+    bytes_received: Arc<AtomicU64>,
+    bytes_sent: Arc<AtomicU64>,
+    xact_time: Arc<AtomicU64>,
+    query_time: Arc<AtomicU64>,
+    wait_time: Arc<AtomicU64>,
+    errors: Arc<AtomicU64>,
+}
+
 /// Internal address stats
 #[derive(Debug, Clone, Default)]
 pub struct AddressStats {
-    pub total_xact_count: Arc<AtomicU64>,
-    pub total_query_count: Arc<AtomicU64>,
-    pub total_received: Arc<AtomicU64>,
-    pub total_sent: Arc<AtomicU64>,
-    pub total_xact_time: Arc<AtomicU64>,
-    pub total_query_time: Arc<AtomicU64>,
-    pub total_wait_time: Arc<AtomicU64>,
-    pub total_errors: Arc<AtomicU64>,
+    total: AddressStatFields,
 
-    pub old_total_xact_count: Arc<AtomicU64>,
-    pub old_total_query_count: Arc<AtomicU64>,
-    pub old_total_received: Arc<AtomicU64>,
-    pub old_total_sent: Arc<AtomicU64>,
-    pub old_total_xact_time: Arc<AtomicU64>,
-    pub old_total_query_time: Arc<AtomicU64>,
-    pub old_total_wait_time: Arc<AtomicU64>,
-    pub old_total_errors: Arc<AtomicU64>,
+    current: AddressStatFields,
 
-    pub avg_query_count: Arc<AtomicU64>,
-    pub avg_query_time: Arc<AtomicU64>,
-    pub avg_recv: Arc<AtomicU64>,
-    pub avg_sent: Arc<AtomicU64>,
-    pub avg_errors: Arc<AtomicU64>,
-    pub avg_xact_time: Arc<AtomicU64>,
-    pub avg_xact_count: Arc<AtomicU64>,
-    pub avg_wait_time: Arc<AtomicU64>,
+    averages: AddressStatFields,
 
     // Determines if the averages have been updated since the last time they were reported
     pub averages_updated: Arc<AtomicBool>,
@@ -43,67 +34,67 @@ impl IntoIterator for AddressStats {
         vec![
             (
                 "total_xact_count".to_string(),
-                self.total_xact_count.load(Ordering::Relaxed),
+                self.total.xact_count.load(Ordering::Relaxed),
             ),
             (
                 "total_query_count".to_string(),
-                self.total_query_count.load(Ordering::Relaxed),
+                self.total.query_count.load(Ordering::Relaxed),
             ),
             (
                 "total_received".to_string(),
-                self.total_received.load(Ordering::Relaxed),
+                self.total.bytes_received.load(Ordering::Relaxed),
             ),
             (
                 "total_sent".to_string(),
-                self.total_sent.load(Ordering::Relaxed),
+                self.total.bytes_sent.load(Ordering::Relaxed),
             ),
             (
                 "total_xact_time".to_string(),
-                self.total_xact_time.load(Ordering::Relaxed),
+                self.total.xact_time.load(Ordering::Relaxed),
             ),
             (
                 "total_query_time".to_string(),
-                self.total_query_time.load(Ordering::Relaxed),
+                self.total.query_time.load(Ordering::Relaxed),
             ),
             (
                 "total_wait_time".to_string(),
-                self.total_wait_time.load(Ordering::Relaxed),
+                self.total.wait_time.load(Ordering::Relaxed),
             ),
             (
                 "total_errors".to_string(),
-                self.total_errors.load(Ordering::Relaxed),
+                self.total.errors.load(Ordering::Relaxed),
             ),
             (
                 "avg_xact_count".to_string(),
-                self.avg_xact_count.load(Ordering::Relaxed),
+                self.averages.xact_count.load(Ordering::Relaxed),
             ),
             (
                 "avg_query_count".to_string(),
-                self.avg_query_count.load(Ordering::Relaxed),
+                self.averages.query_count.load(Ordering::Relaxed),
             ),
             (
                 "avg_recv".to_string(),
-                self.avg_recv.load(Ordering::Relaxed),
+                self.averages.bytes_received.load(Ordering::Relaxed),
             ),
             (
                 "avg_sent".to_string(),
-                self.avg_sent.load(Ordering::Relaxed),
+                self.averages.bytes_sent.load(Ordering::Relaxed),
             ),
             (
                 "avg_errors".to_string(),
-                self.avg_errors.load(Ordering::Relaxed),
+                self.averages.errors.load(Ordering::Relaxed),
             ),
             (
                 "avg_xact_time".to_string(),
-                self.avg_xact_time.load(Ordering::Relaxed),
+                self.averages.xact_time.load(Ordering::Relaxed),
             ),
             (
                 "avg_query_time".to_string(),
-                self.avg_query_time.load(Ordering::Relaxed),
+                self.averages.query_time.load(Ordering::Relaxed),
             ),
             (
                 "avg_wait_time".to_string(),
-                self.avg_wait_time.load(Ordering::Relaxed),
+                self.averages.wait_time.load(Ordering::Relaxed),
             ),
         ]
         .into_iter()
@@ -111,21 +102,120 @@ impl IntoIterator for AddressStats {
 }
 
 impl AddressStats {
+    pub fn xact_count_add(&self) {
+        self.total.xact_count.fetch_add(1, Ordering::Relaxed);
+        self.current.xact_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn query_count_add(&self) {
+        self.total.query_count.fetch_add(1, Ordering::Relaxed);
+        self.current.query_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn bytes_received_add(&self, bytes: u64) {
+        self.total
+            .bytes_received
+            .fetch_add(bytes, Ordering::Relaxed);
+        self.current
+            .bytes_received
+            .fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    pub fn bytes_sent_add(&self, bytes: u64) {
+        self.total.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
+        self.current.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    pub fn xact_time_add(&self, time: u64) {
+        self.total.xact_time.fetch_add(time, Ordering::Relaxed);
+        self.current.xact_time.fetch_add(time, Ordering::Relaxed);
+    }
+
+    pub fn query_time_add(&self, time: u64) {
+        self.total.query_time.fetch_add(time, Ordering::Relaxed);
+        self.current.query_time.fetch_add(time, Ordering::Relaxed);
+    }
+
+    pub fn wait_time_add(&self, time: u64) {
+        self.total.wait_time.fetch_add(time, Ordering::Relaxed);
+        self.current.wait_time.fetch_add(time, Ordering::Relaxed);
+    }
+
     pub fn error(&self) {
-        self.total_errors.fetch_add(1, Ordering::Relaxed);
+        self.total.errors.fetch_add(1, Ordering::Relaxed);
+        self.current.errors.fetch_add(1, Ordering::Relaxed);
     }
 
     pub fn update_averages(&self) {
-        let (totals, averages, old_totals) = self.fields_iterators();
-        for (total, average, old_total) in itertools::izip!(totals, averages, old_totals) {
-            let total_value = total.load(Ordering::Relaxed);
-            let old_total_value = old_total.load(Ordering::Relaxed);
-            average.store(
-                (total_value - old_total_value) / (crate::stats::STAT_PERIOD / 1_000),
-                Ordering::Relaxed,
-            ); // Avg / second
-            old_total.store(total_value, Ordering::Relaxed);
+        let stat_period_per_second = crate::stats::STAT_PERIOD / 1_000;
+
+        // xact_count
+        let current_xact_count = self.current.xact_count.load(Ordering::Relaxed);
+        let current_xact_time = self.current.xact_time.load(Ordering::Relaxed);
+        self.averages.xact_count.store(
+            current_xact_count / stat_period_per_second,
+            Ordering::Relaxed,
+        );
+        if current_xact_count == 0 {
+            self.averages.xact_time.store(0, Ordering::Relaxed);
+        } else {
+            self.averages
+                .xact_time
+                .store(current_xact_time / current_xact_count, Ordering::Relaxed);
         }
+
+        // query_count
+        let current_query_count = self.current.query_count.load(Ordering::Relaxed);
+        let current_query_time = self.current.query_time.load(Ordering::Relaxed);
+        self.averages.query_count.store(
+            current_query_count / stat_period_per_second,
+            Ordering::Relaxed,
+        );
+        if current_query_count == 0 {
+            self.averages.query_time.store(0, Ordering::Relaxed);
+        } else {
+            self.averages
+                .query_time
+                .store(current_query_time / current_query_count, Ordering::Relaxed);
+        }
+
+        // bytes_received
+        let current_bytes_received = self.current.bytes_received.load(Ordering::Relaxed);
+        self.averages.bytes_received.store(
+            current_bytes_received / stat_period_per_second,
+            Ordering::Relaxed,
+        );
+
+        // bytes_sent
+        let current_bytes_sent = self.current.bytes_sent.load(Ordering::Relaxed);
+        self.averages.bytes_sent.store(
+            current_bytes_sent / stat_period_per_second,
+            Ordering::Relaxed,
+        );
+
+        // wait_time
+        let current_wait_time = self.current.wait_time.load(Ordering::Relaxed);
+        self.averages.wait_time.store(
+            current_wait_time / stat_period_per_second,
+            Ordering::Relaxed,
+        );
+
+        // errors
+        let current_errors = self.current.errors.load(Ordering::Relaxed);
+        self.averages
+            .errors
+            .store(current_errors / stat_period_per_second, Ordering::Relaxed);
+    }
+
+    pub fn reset_current_counts(&self) {
+        self.current.xact_count.store(0, Ordering::Relaxed);
+        self.current.xact_time.store(0, Ordering::Relaxed);
+        self.current.query_count.store(0, Ordering::Relaxed);
+        self.current.query_time.store(0, Ordering::Relaxed);
+        self.current.bytes_received.store(0, Ordering::Relaxed);
+        self.current.bytes_sent.store(0, Ordering::Relaxed);
+        self.current.wait_time.store(0, Ordering::Relaxed);
+        self.current.errors.store(0, Ordering::Relaxed);
     }
 
     pub fn populate_row(&self, row: &mut Vec<String>) {
@@ -133,43 +223,4 @@ impl AddressStats {
             row.push(value.to_string());
         }
     }
-
-    fn fields_iterators(
-        &self,
-    ) -> (
-        Vec<Arc<AtomicU64>>,
-        Vec<Arc<AtomicU64>>,
-        Vec<Arc<AtomicU64>>,
-    ) {
-        let mut totals: Vec<Arc<AtomicU64>> = Vec::new();
-        let mut averages: Vec<Arc<AtomicU64>> = Vec::new();
-        let mut old_totals: Vec<Arc<AtomicU64>> = Vec::new();
-
-        totals.push(self.total_xact_count.clone());
-        old_totals.push(self.old_total_xact_count.clone());
-        averages.push(self.avg_xact_count.clone());
-        totals.push(self.total_query_count.clone());
-        old_totals.push(self.old_total_query_count.clone());
-        averages.push(self.avg_query_count.clone());
-        totals.push(self.total_received.clone());
-        old_totals.push(self.old_total_received.clone());
-        averages.push(self.avg_recv.clone());
-        totals.push(self.total_sent.clone());
-        old_totals.push(self.old_total_sent.clone());
-        averages.push(self.avg_sent.clone());
-        totals.push(self.total_xact_time.clone());
-        old_totals.push(self.old_total_xact_time.clone());
-        averages.push(self.avg_xact_time.clone());
-        totals.push(self.total_query_time.clone());
-        old_totals.push(self.old_total_query_time.clone());
-        averages.push(self.avg_query_time.clone());
-        totals.push(self.total_wait_time.clone());
-        old_totals.push(self.old_total_wait_time.clone());
-        averages.push(self.avg_wait_time.clone());
-        totals.push(self.total_errors.clone());
-        old_totals.push(self.old_total_errors.clone());
-        averages.push(self.avg_errors.clone());
-
-        (totals, averages, old_totals)
-    }
 }

src/stats/client.rs

@@ -1,4 +1,3 @@
-use super::PoolStats;
 use super::{get_reporter, Reporter};
 use atomic_enum::atomic_enum;
 use std::sync::atomic::*;
@@ -34,12 +33,16 @@ pub struct ClientStats {
     pool_name: String,
     connect_time: Instant,
 
-    pool_stats: Arc<PoolStats>,
     reporter: Reporter,
 
     /// Total time spent waiting for a connection from pool, measures in microseconds
     pub total_wait_time: Arc<AtomicU64>,
 
+    /// When this client started waiting.
+    /// Stored as microseconds since connect_time so it can fit in an AtomicU64 instead
+    /// of us using an "AtomicInstant"
+    pub wait_start: Arc<AtomicU64>,
+
     /// Current state of the client
     pub state: Arc<AtomicClientState>,
 
@@ -61,8 +64,8 @@ impl Default for ClientStats {
             application_name: String::new(),
             username: String::new(),
             pool_name: String::new(),
-            pool_stats: Arc::new(PoolStats::default()),
             total_wait_time: Arc::new(AtomicU64::new(0)),
+            wait_start: Arc::new(AtomicU64::new(0)),
             state: Arc::new(AtomicClientState::new(ClientState::Idle)),
             transaction_count: Arc::new(AtomicU64::new(0)),
             query_count: Arc::new(AtomicU64::new(0)),
@@ -79,11 +82,9 @@ impl ClientStats {
         username: &str,
         pool_name: &str,
         connect_time: Instant,
-        pool_stats: Arc<PoolStats>,
     ) -> Self {
         Self {
             client_id,
-            pool_stats,
             connect_time,
             application_name: application_name.to_string(),
             username: username.to_string(),
@@ -96,8 +97,6 @@ impl ClientStats {
     /// update metrics on the corresponding pool.
     pub fn disconnect(&self) {
         self.reporter.client_disconnecting(self.client_id);
-        self.pool_stats
-            .client_disconnect(self.state.load(Ordering::Relaxed))
     }
 
     /// Register a client with the stats system. The stats system uses client_id
@@ -105,27 +104,25 @@ impl ClientStats {
     pub fn register(&self, stats: Arc<ClientStats>) {
         self.reporter.client_register(self.client_id, stats);
         self.state.store(ClientState::Idle, Ordering::Relaxed);
-        self.pool_stats.cl_idle.fetch_add(1, Ordering::Relaxed);
     }
 
     /// Reports a client is done querying the server and is no longer assigned a server connection
     pub fn idle(&self) {
-        self.pool_stats
-            .client_idle(self.state.load(Ordering::Relaxed));
         self.state.store(ClientState::Idle, Ordering::Relaxed);
     }
 
     /// Reports a client is waiting for a connection
     pub fn waiting(&self) {
-        self.pool_stats
-            .client_waiting(self.state.load(Ordering::Relaxed));
+        // safe to truncate, we only lose info if duration is greater than ~585,000 years
+        self.wait_start.store(
+            Instant::now().duration_since(self.connect_time).as_micros() as u64,
+            Ordering::Relaxed,
+        );
         self.state.store(ClientState::Waiting, Ordering::Relaxed);
     }
 
     /// Reports a client is done waiting for a connection and is about to query the server.
     pub fn active(&self) {
-        self.pool_stats
-            .client_active(self.state.load(Ordering::Relaxed));
         self.state.store(ClientState::Active, Ordering::Relaxed);
     }
 

src/stats/pool.rs

@@ -1,36 +1,139 @@
-use crate::config::Pool;
-use crate::config::PoolMode;
-use crate::pool::PoolIdentifier;
-use std::sync::atomic::*;
-use std::sync::Arc;
+use log::debug;
 
-use super::get_reporter;
-use super::Reporter;
 use super::{ClientState, ServerState};
+use crate::{config::PoolMode, messages::DataType, pool::PoolIdentifier};
+use std::collections::HashMap;
+use std::sync::atomic::*;
+use tokio::time::Instant;
 
-#[derive(Debug, Clone, Default)]
+use crate::pool::get_all_pools;
+
+#[derive(Debug, Clone)]
 /// A struct that holds information about a Pool .
 pub struct PoolStats {
-    // Pool identifier, cannot be changed after creating the instance
-    identifier: PoolIdentifier,
+    pub identifier: PoolIdentifier,
+    pub mode: PoolMode,
+    pub cl_idle: u64,
+    pub cl_active: u64,
+    pub cl_waiting: u64,
+    pub cl_cancel_req: u64,
+    pub sv_active: u64,
+    pub sv_idle: u64,
+    pub sv_used: u64,
+    pub sv_tested: u64,
+    pub sv_login: u64,
+    pub maxwait: u64,
+}
+impl PoolStats {
+    pub fn new(identifier: PoolIdentifier, mode: PoolMode) -> Self {
+        PoolStats {
+            identifier,
+            mode,
+            cl_idle: 0,
+            cl_active: 0,
+            cl_waiting: 0,
+            cl_cancel_req: 0,
+            sv_active: 0,
+            sv_idle: 0,
+            sv_used: 0,
+            sv_tested: 0,
+            sv_login: 0,
+            maxwait: 0,
+        }
+    }
 
-    // Pool Config, cannot be changed after creating the instance
-    config: Pool,
+    pub fn construct_pool_lookup() -> HashMap<PoolIdentifier, PoolStats> {
+        let mut map: HashMap<PoolIdentifier, PoolStats> = HashMap::new();
+        let client_map = super::get_client_stats();
+        let server_map = super::get_server_stats();
 
-    // A reference to the global reporter.
-    reporter: Reporter,
+        for (identifier, pool) in get_all_pools() {
+            map.insert(
+                identifier.clone(),
+                PoolStats::new(identifier, pool.settings.pool_mode),
+            );
+        }
 
-    /// Counters (atomics)
-    pub cl_idle: Arc<AtomicU64>,
-    pub cl_active: Arc<AtomicU64>,
-    pub cl_waiting: Arc<AtomicU64>,
-    pub cl_cancel_req: Arc<AtomicU64>,
-    pub sv_active: Arc<AtomicU64>,
-    pub sv_idle: Arc<AtomicU64>,
-    pub sv_used: Arc<AtomicU64>,
-    pub sv_tested: Arc<AtomicU64>,
-    pub sv_login: Arc<AtomicU64>,
-    pub maxwait: Arc<AtomicU64>,
+        let now = Instant::now();
+        for client in client_map.values() {
+            match map.get_mut(&PoolIdentifier {
+                db: client.pool_name(),
+                user: client.username(),
+            }) {
+                Some(pool_stats) => {
+                    match client.state.load(Ordering::Relaxed) {
+                        ClientState::Active => pool_stats.cl_active += 1,
+                        ClientState::Idle => pool_stats.cl_idle += 1,
+                        ClientState::Waiting => {
+                            pool_stats.cl_waiting += 1;
+                            // wait_start is measured as microseconds since connect_time
+                            // so compute wait_time as (now() - connect_time) - (wait_start - connect_time)
+                            let duration_since_connect = now.duration_since(client.connect_time());
+                            let wait_time = (duration_since_connect.as_micros() as u64)
+                                - client.wait_start.load(Ordering::Relaxed);
+                            pool_stats.maxwait = std::cmp::max(pool_stats.maxwait, wait_time);
+                        }
+                    }
+                }
+                None => debug!("Client from an obselete pool"),
+            }
+        }
+
+        for server in server_map.values() {
+            match map.get_mut(&PoolIdentifier {
+                db: server.pool_name(),
+                user: server.username(),
+            }) {
+                Some(pool_stats) => match server.state.load(Ordering::Relaxed) {
+                    ServerState::Active => pool_stats.sv_active += 1,
+                    ServerState::Idle => pool_stats.sv_idle += 1,
+                    ServerState::Login => pool_stats.sv_login += 1,
+                    ServerState::Tested => pool_stats.sv_tested += 1,
+                },
+                None => debug!("Server from an obselete pool"),
+            }
+        }
+
+        map
+    }
+
+    pub fn generate_header() -> Vec<(&'static str, DataType)> {
+        vec![
+            ("database", DataType::Text),
+            ("user", DataType::Text),
+            ("pool_mode", DataType::Text),
+            ("cl_idle", DataType::Numeric),
+            ("cl_active", DataType::Numeric),
+            ("cl_waiting", DataType::Numeric),
+            ("cl_cancel_req", DataType::Numeric),
+            ("sv_active", DataType::Numeric),
+            ("sv_idle", DataType::Numeric),
+            ("sv_used", DataType::Numeric),
+            ("sv_tested", DataType::Numeric),
+            ("sv_login", DataType::Numeric),
+            ("maxwait", DataType::Numeric),
+            ("maxwait_us", DataType::Numeric),
+        ]
+    }
+
+    pub fn generate_row(&self) -> Vec<String> {
+        vec![
+            self.identifier.db.clone(),
+            self.identifier.user.clone(),
+            self.mode.to_string(),
+            self.cl_idle.to_string(),
+            self.cl_active.to_string(),
+            self.cl_waiting.to_string(),
+            self.cl_cancel_req.to_string(),
+            self.sv_active.to_string(),
+            self.sv_idle.to_string(),
+            self.sv_used.to_string(),
+            self.sv_tested.to_string(),
+            self.sv_login.to_string(),
+            (self.maxwait / 1_000_000).to_string(),
+            (self.maxwait % 1_000_000).to_string(),
+        ]
+    }
 }
 
 impl IntoIterator for PoolStats {
@@ -39,236 +142,18 @@ impl IntoIterator for PoolStats {
 
     fn into_iter(self) -> Self::IntoIter {
         vec![
-            ("cl_idle".to_string(), self.cl_idle.load(Ordering::Relaxed)),
-            (
-                "cl_active".to_string(),
-                self.cl_active.load(Ordering::Relaxed),
-            ),
-            (
-                "cl_waiting".to_string(),
-                self.cl_waiting.load(Ordering::Relaxed),
-            ),
-            (
-                "cl_cancel_req".to_string(),
-                self.cl_cancel_req.load(Ordering::Relaxed),
-            ),
-            (
-                "sv_active".to_string(),
-                self.sv_active.load(Ordering::Relaxed),
-            ),
-            ("sv_idle".to_string(), self.sv_idle.load(Ordering::Relaxed)),
-            ("sv_used".to_string(), self.sv_used.load(Ordering::Relaxed)),
-            (
-                "sv_tested".to_string(),
-                self.sv_tested.load(Ordering::Relaxed),
-            ),
-            (
-                "sv_login".to_string(),
-                self.sv_login.load(Ordering::Relaxed),
-            ),
-            (
-                "maxwait".to_string(),
-                self.maxwait.load(Ordering::Relaxed) / 1_000_000,
-            ),
-            (
-                "maxwait_us".to_string(),
-                self.maxwait.load(Ordering::Relaxed) % 1_000_000,
-            ),
+            ("cl_idle".to_string(), self.cl_idle),
+            ("cl_active".to_string(), self.cl_active),
+            ("cl_waiting".to_string(), self.cl_waiting),
+            ("cl_cancel_req".to_string(), self.cl_cancel_req),
+            ("sv_active".to_string(), self.sv_active),
+            ("sv_idle".to_string(), self.sv_idle),
+            ("sv_used".to_string(), self.sv_used),
+            ("sv_tested".to_string(), self.sv_tested),
+            ("sv_login".to_string(), self.sv_login),
+            ("maxwait".to_string(), self.maxwait / 1_000_000),
+            ("maxwait_us".to_string(), self.maxwait % 1_000_000),
         ]
         .into_iter()
     }
 }
-
-impl PoolStats {
-    pub fn new(identifier: PoolIdentifier, config: Pool) -> Self {
-        Self {
-            identifier,
-            config,
-            reporter: get_reporter(),
-            ..Default::default()
-        }
-    }
-
-    // Getters
-    pub fn register(&self, stats: Arc<PoolStats>) {
-        self.reporter.pool_register(self.identifier.clone(), stats);
-    }
-
-    pub fn database(&self) -> String {
-        self.identifier.db.clone()
-    }
-
-    pub fn user(&self) -> String {
-        self.identifier.user.clone()
-    }
-
-    pub fn pool_mode(&self) -> PoolMode {
-        self.config.pool_mode
-    }
-
-    /// Populates an array of strings with counters (used by admin in show pools)
-    pub fn populate_row(&self, row: &mut Vec<String>) {
-        for (_key, value) in self.clone() {
-            row.push(value.to_string());
-        }
-    }
-
-    /// Deletes the maxwait counter, this is done everytime we obtain metrics
-    pub fn clear_maxwait(&self) {
-        self.maxwait.store(0, Ordering::Relaxed);
-    }
-
-    /// Notified when a server of the pool enters login state.
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the server that notifies.
-    pub fn server_login(&self, from: ServerState) {
-        self.sv_login.fetch_add(1, Ordering::Relaxed);
-        if from != ServerState::Login {
-            self.decrease_from_server_state(from);
-        }
-    }
-
-    /// Notified when a server of the pool become 'active'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the server that notifies.
-    pub fn server_active(&self, from: ServerState) {
-        self.sv_active.fetch_add(1, Ordering::Relaxed);
-        if from != ServerState::Active {
-            self.decrease_from_server_state(from);
-        }
-    }
-
-    /// Notified when a server of the pool become 'tested'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the server that notifies.
-    pub fn server_tested(&self, from: ServerState) {
-        self.sv_tested.fetch_add(1, Ordering::Relaxed);
-        if from != ServerState::Tested {
-            self.decrease_from_server_state(from);
-        }
-    }
-
-    /// Notified when a server of the pool become 'idle'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the server that notifies.
-    pub fn server_idle(&self, from: ServerState) {
-        self.sv_idle.fetch_add(1, Ordering::Relaxed);
-        if from != ServerState::Idle {
-            self.decrease_from_server_state(from);
-        }
-    }
-
-    /// Notified when a client of the pool become 'waiting'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the client that notifies.
-    pub fn client_waiting(&self, from: ClientState) {
-        if from != ClientState::Waiting {
-            self.cl_waiting.fetch_add(1, Ordering::Relaxed);
-            self.decrease_from_client_state(from);
-        }
-    }
-
-    /// Notified when a client of the pool become 'active'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the client that notifies.
-    pub fn client_active(&self, from: ClientState) {
-        if from != ClientState::Active {
-            self.cl_active.fetch_add(1, Ordering::Relaxed);
-            self.decrease_from_client_state(from);
-        }
-    }
-
-    /// Notified when a client of the pool become 'idle'
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the client that notifies.
-    pub fn client_idle(&self, from: ClientState) {
-        if from != ClientState::Idle {
-            self.cl_idle.fetch_add(1, Ordering::Relaxed);
-            self.decrease_from_client_state(from);
-        }
-    }
-
-    /// Notified when a client disconnects.
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the client that notifies.
-    pub fn client_disconnect(&self, from: ClientState) {
-        let counter = match from {
-            ClientState::Idle => &self.cl_idle,
-            ClientState::Waiting => &self.cl_waiting,
-            ClientState::Active => &self.cl_active,
-        };
-
-        Self::decrease_counter(counter.clone());
-    }
-
-    /// Notified when a server disconnects.
-    ///
-    /// Arguments:
-    ///
-    /// `from`: The state of the client that notifies.
-    pub fn server_disconnect(&self, from: ServerState) {
-        let counter = match from {
-            ServerState::Active => &self.sv_active,
-            ServerState::Idle => &self.sv_idle,
-            ServerState::Login => &self.sv_login,
-            ServerState::Tested => &self.sv_tested,
-        };
-        Self::decrease_counter(counter.clone());
-    }
-
-    // helpers for counter decrease
-    fn decrease_from_server_state(&self, from: ServerState) {
-        let counter = match from {
-            ServerState::Tested => &self.sv_tested,
-            ServerState::Active => &self.sv_active,
-            ServerState::Idle => &self.sv_idle,
-            ServerState::Login => &self.sv_login,
-        };
-        Self::decrease_counter(counter.clone());
-    }
-
-    fn decrease_from_client_state(&self, from: ClientState) {
-        let counter = match from {
-            ClientState::Active => &self.cl_active,
-            ClientState::Idle => &self.cl_idle,
-            ClientState::Waiting => &self.cl_waiting,
-        };
-        Self::decrease_counter(counter.clone());
-    }
-
-    fn decrease_counter(value: Arc<AtomicU64>) {
-        if value.load(Ordering::Relaxed) > 0 {
-            value.fetch_sub(1, Ordering::Relaxed);
-        }
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-
-    #[test]
-    fn test_decrease() {
-        let stat: PoolStats = PoolStats::default();
-        stat.server_login(ServerState::Login);
-        stat.server_idle(ServerState::Login);
-        assert_eq!(stat.sv_login.load(Ordering::Relaxed), 0);
-        assert_eq!(stat.sv_idle.load(Ordering::Relaxed), 1);
-    }
-}

src/stats/server.rs

@@ -1,5 +1,4 @@
 use super::AddressStats;
-use super::PoolStats;
 use super::{get_reporter, Reporter};
 use crate::config::Address;
 use atomic_enum::atomic_enum;
@@ -38,7 +37,6 @@ pub struct ServerStats {
     address: Address,
     connect_time: Instant,
 
-    pool_stats: Arc<PoolStats>,
     reporter: Reporter,
 
     /// Data
@@ -49,6 +47,10 @@ pub struct ServerStats {
     pub transaction_count: Arc<AtomicU64>,
     pub query_count: Arc<AtomicU64>,
     pub error_count: Arc<AtomicU64>,
+    pub prepared_hit_count: Arc<AtomicU64>,
+    pub prepared_miss_count: Arc<AtomicU64>,
+    pub prepared_eviction_count: Arc<AtomicU64>,
+    pub prepared_cache_size: Arc<AtomicU64>,
 }
 
 impl Default for ServerStats {
@@ -57,7 +59,6 @@ impl Default for ServerStats {
             server_id: 0,
             application_name: Arc::new(RwLock::new(String::new())),
             address: Address::default(),
-            pool_stats: Arc::new(PoolStats::default()),
             connect_time: Instant::now(),
             state: Arc::new(AtomicServerState::new(ServerState::Login)),
             bytes_sent: Arc::new(AtomicU64::new(0)),
@@ -66,15 +67,18 @@ impl Default for ServerStats {
             query_count: Arc::new(AtomicU64::new(0)),
             error_count: Arc::new(AtomicU64::new(0)),
             reporter: get_reporter(),
+            prepared_hit_count: Arc::new(AtomicU64::new(0)),
+            prepared_miss_count: Arc::new(AtomicU64::new(0)),
+            prepared_eviction_count: Arc::new(AtomicU64::new(0)),
+            prepared_cache_size: Arc::new(AtomicU64::new(0)),
         }
     }
 }
 
 impl ServerStats {
-    pub fn new(address: Address, pool_stats: Arc<PoolStats>, connect_time: Instant) -> Self {
+    pub fn new(address: Address, connect_time: Instant) -> Self {
         Self {
             address,
-            pool_stats,
             connect_time,
             server_id: rand::random::<i32>(),
             ..Default::default()
@@ -96,9 +100,6 @@ impl ServerStats {
     /// Reports a server connection is no longer assigned to a client
     /// and is available for the next client to pick it up
     pub fn idle(&self) {
-        self.pool_stats
-            .server_idle(self.state.load(Ordering::Relaxed));
-
         self.state.store(ServerState::Idle, Ordering::Relaxed);
     }
 
@@ -106,22 +107,16 @@ impl ServerStats {
     /// Also updates metrics on the pool regarding server usage.
     pub fn disconnect(&self) {
         self.reporter.server_disconnecting(self.server_id);
-        self.pool_stats
-            .server_disconnect(self.state.load(Ordering::Relaxed))
     }
 
     /// Reports a server connection is being tested before being given to a client.
     pub fn tested(&self) {
         self.set_undefined_application();
-        self.pool_stats
-            .server_tested(self.state.load(Ordering::Relaxed));
         self.state.store(ServerState::Tested, Ordering::Relaxed);
     }
 
     /// Reports a server connection is attempting to login.
     pub fn login(&self) {
-        self.pool_stats
-            .server_login(self.state.load(Ordering::Relaxed));
         self.state.store(ServerState::Login, Ordering::Relaxed);
         self.set_undefined_application();
     }
@@ -129,8 +124,6 @@ impl ServerStats {
     /// Reports a server connection has been assigned to a client that
     /// is about to query the server
     pub fn active(&self, application_name: String) {
-        self.pool_stats
-            .server_active(self.state.load(Ordering::Relaxed));
         self.state.store(ServerState::Active, Ordering::Relaxed);
         self.set_application(application_name);
     }
@@ -152,11 +145,11 @@ impl ServerStats {
 
     // Helper methods for show_servers
     pub fn pool_name(&self) -> String {
-        self.pool_stats.database()
+        self.address.pool_name.clone()
     }
 
     pub fn username(&self) -> String {
-        self.pool_stats.user()
+        self.address.username.clone()
     }
 
     pub fn address_name(&self) -> String {
@@ -177,27 +170,17 @@ impl ServerStats {
     }
 
     pub fn checkout_time(&self, microseconds: u64, application_name: String) {
-        // Update server stats and address aggergation stats
+        // Update server stats and address aggregation stats
         self.set_application(application_name);
-        self.address
-            .stats
-            .total_wait_time
-            .fetch_add(microseconds, Ordering::Relaxed);
-        self.pool_stats
-            .maxwait
-            .fetch_max(microseconds, Ordering::Relaxed);
+        self.address.stats.wait_time_add(microseconds);
     }
 
     /// Report a query executed by a client against a server
     pub fn query(&self, milliseconds: u64, application_name: &str) {
         self.set_application(application_name.to_string());
-        let address_stats = self.address_stats();
-        address_stats
-            .total_query_count
-            .fetch_add(1, Ordering::Relaxed);
-        address_stats
-            .total_query_time
-            .fetch_add(milliseconds, Ordering::Relaxed);
+        self.address.stats.query_count_add();
+        self.address.stats.query_time_add(milliseconds);
+        self.query_count.fetch_add(1, Ordering::Relaxed);
     }
 
     /// Report a transaction executed by a client a server
@@ -208,29 +191,39 @@ impl ServerStats {
         self.set_application(application_name.to_string());
 
         self.transaction_count.fetch_add(1, Ordering::Relaxed);
-        self.address
-            .stats
-            .total_xact_count
-            .fetch_add(1, Ordering::Relaxed);
+        self.address.stats.xact_count_add();
     }
 
     /// Report data sent to a server
     pub fn data_sent(&self, amount_bytes: usize) {
         self.bytes_sent
             .fetch_add(amount_bytes as u64, Ordering::Relaxed);
-        self.address
-            .stats
-            .total_sent
-            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
+        self.address.stats.bytes_sent_add(amount_bytes as u64);
     }
 
     /// Report data received from a server
     pub fn data_received(&self, amount_bytes: usize) {
         self.bytes_received
             .fetch_add(amount_bytes as u64, Ordering::Relaxed);
-        self.address
-            .stats
-            .total_received
-            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
+        self.address.stats.bytes_received_add(amount_bytes as u64);
+    }
+
+    /// Report a prepared statement that already exists on the server.
+    pub fn prepared_cache_hit(&self) {
+        self.prepared_hit_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Report a prepared statement that does not exist on the server yet.
+    pub fn prepared_cache_miss(&self) {
+        self.prepared_miss_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn prepared_cache_add(&self) {
+        self.prepared_cache_size.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn prepared_cache_remove(&self) {
+        self.prepared_eviction_count.fetch_add(1, Ordering::Relaxed);
+        self.prepared_cache_size.fetch_sub(1, Ordering::Relaxed);
     }
 }

tests/docker/Dockerfile

@@ -1,8 +1,13 @@
 FROM rust:bullseye
 
+COPY --from=sclevine/yj /bin/yj /bin/yj
+RUN /bin/yj -h
 RUN apt-get update && apt-get install llvm-11 psmisc postgresql-contrib postgresql-client ruby ruby-dev libpq-dev python3 python3-pip lcov curl sudo iproute2 -y
 RUN cargo install cargo-binutils rustfilt
 RUN rustup component add llvm-tools-preview
 RUN sudo gem install bundler
 RUN wget -O toxiproxy-2.4.0.deb https://github.com/Shopify/toxiproxy/releases/download/v2.4.0/toxiproxy_2.4.0_linux_$(dpkg --print-architecture).deb && \
 	sudo dpkg -i toxiproxy-2.4.0.deb
+RUN wget -O go1.21.3.linux-$(dpkg --print-architecture).tar.gz https://go.dev/dl/go1.21.3.linux-$(dpkg --print-architecture).tar.gz && \
+    sudo tar -C /usr/local -xzf go1.21.3.linux-$(dpkg --print-architecture).tar.gz && \
+    rm go1.21.3.linux-$(dpkg --print-architecture).tar.gz

tests/go/go.mod

@@ -0,0 +1,5 @@
+module pgcat
+
+go 1.21
+
+require github.com/lib/pq v1.10.9

tests/go/go.sum

@@ -0,0 +1,2 @@
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=

tests/go/pgcat.toml

@@ -0,0 +1,162 @@
+#
+# PgCat config example.
+#
+
+#
+# General pooler settings
+[general]
+# What IP to run on, 0.0.0.0 means accessible from everywhere.
+host = "0.0.0.0"
+
+# Port to run on, same as PgBouncer used in this example.
+port = "${PORT}"
+
+# Whether to enable prometheus exporter or not.
+enable_prometheus_exporter = true
+
+# Port at which prometheus exporter listens on.
+prometheus_exporter_port = 9930
+
+# How long to wait before aborting a server connection (ms).
+connect_timeout = 1000
+
+# How much time to give the health check query to return with a result (ms).
+healthcheck_timeout = 1000
+
+# How long to keep connection available for immediate re-use, without running a healthcheck query on it
+healthcheck_delay = 30000
+
+# How much time to give clients during shutdown before forcibly killing client connections (ms).
+shutdown_timeout = 5000
+
+# For how long to ban a server if it fails a health check (seconds).
+ban_time = 60 # Seconds
+
+# If we should log client connections
+log_client_connections = false
+
+# If we should log client disconnections
+log_client_disconnections = false
+
+# Reload config automatically if it changes.
+autoreload = 15000
+
+server_round_robin = false
+
+# TLS
+tls_certificate = "../../.circleci/server.cert"
+tls_private_key = "../../.circleci/server.key"
+
+# Credentials to access the virtual administrative database (pgbouncer or pgcat)
+# Connecting to that database allows running commands like `SHOW POOLS`, `SHOW DATABASES`, etc..
+admin_username = "admin_user"
+admin_password = "admin_pass"
+
+# pool
+# configs are structured as pool.<pool_name>
+# the pool_name is what clients use as database name when connecting
+# For the example below a client can connect using "postgres://sharding_user:sharding_user@pgcat_host:pgcat_port/sharded_db"
+[pools.sharded_db]
+# Pool mode (see PgBouncer docs for more).
+# session: one server connection per connected client
+# transaction: one server connection per client transaction
+pool_mode = "transaction"
+
+# If the client doesn't specify, route traffic to
+# this role by default.
+#
+# any: round-robin between primary and replicas,
+# replica: round-robin between replicas only without touching the primary,
+# primary: all queries go to the primary unless otherwise specified.
+default_role = "any"
+
+# Query parser. If enabled, we'll attempt to parse
+# every incoming query to determine if it's a read or a write.
+# If it's a read query, we'll direct it to a replica. Otherwise, if it's a write,
+# we'll direct it to the primary.
+query_parser_enabled = true
+
+# If the query parser is enabled and this setting is enabled, we'll attempt to
+# infer the role from the query itself.
+query_parser_read_write_splitting = true
+
+# If the query parser is enabled and this setting is enabled, the primary will be part of the pool of databases used for
+# load balancing of read queries. Otherwise, the primary will only be used for write
+# queries. The primary can always be explicitely selected with our custom protocol.
+primary_reads_enabled = true
+
+# So what if you wanted to implement a different hashing function,
+# or you've already built one and you want this pooler to use it?
+#
+# Current options:
+#
+# pg_bigint_hash: PARTITION BY HASH (Postgres hashing function)
+# sha1: A hashing function based on SHA1
+#
+sharding_function = "pg_bigint_hash"
+
+# Prepared statements cache size.
+prepared_statements_cache_size = 500
+
+# Credentials for users that may connect to this cluster
+[pools.sharded_db.users.0]
+username = "sharding_user"
+password = "sharding_user"
+# Maximum number of server connections that can be established for this user
+# The maximum number of connection from a single Pgcat process to any database in the cluster
+# is the sum of pool_size across all users.
+pool_size = 5
+statement_timeout = 0
+
+
+[pools.sharded_db.users.1]
+username = "other_user"
+password = "other_user"
+pool_size = 21
+statement_timeout = 30000
+
+# Shard 0
+[pools.sharded_db.shards.0]
+# [ host, port, role ]
+servers = [
+    [ "127.0.0.1", 5432, "primary" ],
+    [ "localhost", 5432, "replica" ]
+]
+# Database name (e.g. "postgres")
+database = "shard0"
+
+[pools.sharded_db.shards.1]
+servers = [
+    [ "127.0.0.1", 5432, "primary" ],
+    [ "localhost", 5432, "replica" ],
+]
+database = "shard1"
+
+[pools.sharded_db.shards.2]
+servers = [
+    [ "127.0.0.1", 5432, "primary" ],
+    [ "localhost", 5432, "replica" ],
+]
+database = "shard2"
+
+
+[pools.simple_db]
+pool_mode = "session"
+default_role = "primary"
+query_parser_enabled = true
+query_parser_read_write_splitting = true
+primary_reads_enabled = true
+sharding_function = "pg_bigint_hash"
+
+[pools.simple_db.users.0]
+username = "simple_user"
+password = "simple_user"
+pool_size = 5
+statement_timeout = 30000
+
+[pools.simple_db.shards.0]
+servers = [
+    [ "127.0.0.1", 5432, "primary" ],
+    [ "localhost", 5432, "replica" ]
+]
+database = "some_db"

tests/go/prepared_test.go

@@ -0,0 +1,52 @@
+package pgcat
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	_ "github.com/lib/pq"
+	"testing"
+)
+
+func Test(t *testing.T) {
+	t.Cleanup(setup(t))
+	t.Run("Named parameterized prepared statement works", namedParameterizedPreparedStatement)
+	t.Run("Unnamed parameterized prepared statement works", unnamedParameterizedPreparedStatement)
+}
+
+func namedParameterizedPreparedStatement(t *testing.T) {
+	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d database=sharded_db user=sharding_user password=sharding_user sslmode=disable", port))
+	if err != nil {
+		t.Fatalf("could not open connection: %+v", err)
+	}
+
+	stmt, err := db.Prepare("SELECT $1")
+
+	if err != nil {
+		t.Fatalf("could not prepare: %+v", err)
+	}
+
+	for i := 0; i < 100; i++ {
+		rows, err := stmt.Query(1)
+		if err != nil {
+			t.Fatalf("could not query: %+v", err)
+		}
+		_ = rows.Close()
+	}
+}
+
+func unnamedParameterizedPreparedStatement(t *testing.T) {
+	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d database=sharded_db user=sharding_user password=sharding_user sslmode=disable", port))
+	if err != nil {
+		t.Fatalf("could not open connection: %+v", err)
+	}
+
+	for i := 0; i < 100; i++ {
+		// Under the hood QueryContext generates an unnamed parameterized prepared statement
+		rows, err := db.QueryContext(context.Background(), "SELECT $1", 1)
+		if err != nil {
+			t.Fatalf("could not query: %+v", err)
+		}
+		_ = rows.Close()
+	}
+}

tests/go/setup.go

@@ -0,0 +1,81 @@
+package pgcat
+
+import (
+	"context"
+	"database/sql"
+	_ "embed"
+	"fmt"
+	"math/rand"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+)
+
+//go:embed pgcat.toml
+var pgcatCfg string
+
+var port = rand.Intn(32760-20000) + 20000
+
+func setup(t *testing.T) func() {
+	cfg, err := os.CreateTemp("/tmp", "pgcat_cfg_*.toml")
+	if err != nil {
+		t.Fatalf("could not create temp file: %+v", err)
+	}
+
+	pgcatCfg = strings.Replace(pgcatCfg, "\"${PORT}\"", fmt.Sprintf("%d", port), 1)
+
+	_, err = cfg.Write([]byte(pgcatCfg))
+	if err != nil {
+		t.Fatalf("could not write temp file: %+v", err)
+	}
+
+	commandPath := "../../target/debug/pgcat"
+	if os.Getenv("CARGO_TARGET_DIR") != "" {
+		commandPath = os.Getenv("CARGO_TARGET_DIR") + "/debug/pgcat"
+	}
+
+	cmd := exec.Command(commandPath, cfg.Name())
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	go func() {
+		err = cmd.Run()
+		if err != nil {
+			t.Errorf("could not run pgcat: %+v", err)
+		}
+	}()
+
+	deadline, cancelFunc := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
+	defer cancelFunc()
+	for {
+		select {
+		case <-deadline.Done():
+			break
+		case <-time.After(50 * time.Millisecond):
+			db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d database=pgcat user=admin_user password=admin_pass sslmode=disable", port))
+			if err != nil {
+				continue
+			}
+			rows, err := db.QueryContext(deadline, "SHOW STATS")
+			if err != nil {
+				continue
+			}
+			_ = rows.Close()
+			_ = db.Close()
+			break
+		}
+		break
+	}
+
+	return func() {
+		err := cmd.Process.Signal(os.Interrupt)
+		if err != nil {
+			t.Fatalf("could not interrupt pgcat: %+v", err)
+		}
+		err = os.Remove(cfg.Name())
+		if err != nil {
+			t.Fatalf("could not remove temp file: %+v", err)
+		}
+	}
+}

tests/pgbench/simple.sql

@@ -36,4 +36,4 @@ SELECT abalance FROM pgbench_accounts WHERE aid = :aid;
 SET SERVER ROLE TO 'replica';
 
 -- Read load balancing
-SELECT abalance FROM pgbench_accounts WHERE aid = :aid;
+SELECT abalance FROM pgbench_accounts WHERE aid = :aid;

tests/python/tests.py

@@ -63,6 +63,7 @@ def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.
 
 
 def test_normal_db_access():
+    pgcat_start()
     conn, cur = connect_db(autocommit=False)
     cur.execute("SELECT 1")
     res = cur.fetchall()

tests/ruby/admin_spec.rb

@@ -11,326 +11,6 @@ describe "Admin" do
     processes.pgcat.shutdown
   end
 
-  describe "SHOW STATS" do
-    context "clients connect and make one query" do
-      it "updates *_query_time and *_wait_time" do
-        connections = Array.new(3) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          Thread.new { c.async_exec("SELECT pg_sleep(0.25)") }
-        end
-        sleep(1)
-        connections.map(&:close)
-
-        # wait for averages to be calculated, we shouldn't do this too often
-        sleep(15.5)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW STATS")[0]
-        admin_conn.close
-        expect(results["total_query_time"].to_i).to be_within(200).of(750)
-        expect(results["avg_query_time"].to_i).to be_within(20).of(50)
-
-        expect(results["total_wait_time"].to_i).to_not eq(0)
-        expect(results["avg_wait_time"].to_i).to_not eq(0)
-      end
-    end
-  end
-
-  describe "SHOW POOLS" do
-    context "bad credentials" do
-      it "does not change any stats" do
-        bad_password_url = URI(pgcat_conn_str)
-        bad_password_url.password = "wrong"
-        expect { PG::connect("#{bad_password_url.to_s}?application_name=bad_password") }.to raise_error(PG::ConnectionBad)
-
-        sleep(1)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-
-        expect(results["sv_idle"]).to eq("1")
-      end
-    end
-
-    context "bad database name" do
-      it "does not change any stats" do
-        bad_db_url = URI(pgcat_conn_str)
-        bad_db_url.path = "/wrong_db"
-        expect { PG::connect("#{bad_db_url.to_s}?application_name=bad_db") }.to raise_error(PG::ConnectionBad)
-
-        sleep(1)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-
-        expect(results["sv_idle"]).to eq("1")
-      end
-    end
-
-    context "client connects but issues no queries" do
-      it "only affects cl_idle stats" do
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-
-        before_test = admin_conn.async_exec("SHOW POOLS")[0]["sv_idle"]
-        connections = Array.new(20) { PG::connect(pgcat_conn_str) }
-        sleep(1)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["cl_idle"]).to eq("20")
-        expect(results["sv_idle"]).to eq(before_test)
-
-        connections.map(&:close)
-        sleep(1.1)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_active cl_idle cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["sv_idle"]).to eq(before_test)
-      end
-    end
-
-    context "clients connect and make one query" do
-      it "only affects cl_idle, sv_idle stats" do
-        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          Thread.new { c.async_exec("SELECT pg_sleep(2.5)") }
-        end
-
-        sleep(1.1)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_waiting cl_cancel_req sv_idle sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["cl_active"]).to eq("5")
-        expect(results["sv_active"]).to eq("5")
-
-        sleep(3)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["cl_idle"]).to eq("5")
-        expect(results["sv_idle"]).to eq("5")
-
-        connections.map(&:close)
-        sleep(1)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["sv_idle"]).to eq("5")
-      end
-    end
-
-    context "client connects and opens a transaction and closes connection uncleanly" do
-      it "produces correct statistics" do
-        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          Thread.new do
-            c.async_exec("BEGIN")
-            c.async_exec("SELECT pg_sleep(0.01)")
-            c.close
-          end
-        end
-
-        sleep(1.1)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["sv_idle"]).to eq("5")
-      end
-    end
-
-    context "client fail to checkout connection from the pool" do
-      it "counts clients as idle" do
-        new_configs = processes.pgcat.current_config
-        new_configs["general"]["connect_timeout"] = 500
-        new_configs["general"]["ban_time"] = 1
-        new_configs["general"]["shutdown_timeout"] = 1
-        new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
-        processes.pgcat.update_config(new_configs)
-        processes.pgcat.reload_config
-
-        threads = []
-        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          threads << Thread.new { c.async_exec("SELECT pg_sleep(1)") rescue PG::SystemError }
-        end
-
-        sleep(2)
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["cl_idle"]).to eq("5")
-        expect(results["sv_idle"]).to eq("1")
-
-        threads.map(&:join)
-        connections.map(&:close)
-      end
-    end
-
-    context "clients connects and disconnect normally" do
-      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
-
-      it 'shows the same number of clients before and after' do
-        clients_before = clients_connected_to_pool(processes: processes)
-        threads = []
-        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          threads << Thread.new { c.async_exec("SELECT 1") }
-        end
-        clients_between = clients_connected_to_pool(processes: processes)
-        expect(clients_before).not_to eq(clients_between)
-        connections.each(&:close)
-        clients_after = clients_connected_to_pool(processes: processes)
-        expect(clients_before).to eq(clients_after)
-      end
-    end
-
-    context "clients connects and disconnect abruptly" do
-      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 10) }
-
-      it 'shows the same number of clients before and after' do
-        threads = []
-        connections = Array.new(2) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          threads << Thread.new { c.async_exec("SELECT 1") }
-        end
-        clients_before = clients_connected_to_pool(processes: processes)
-        random_string = (0...8).map { (65 + rand(26)).chr }.join
-        connection_string = "#{pgcat_conn_str}?application_name=#{random_string}"
-        faulty_client = Process.spawn("psql -Atx #{connection_string} >/dev/null")
-        sleep(1)
-        # psql starts two processes, we only know the pid of the parent, this
-        # ensure both are killed
-        `pkill -9 -f '#{random_string}'`
-        Process.wait(faulty_client)
-        clients_after = clients_connected_to_pool(processes: processes)
-        expect(clients_before).to eq(clients_after)
-      end
-    end
-
-    context "clients overwhelm server pools" do
-      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
-
-      it "cl_waiting is updated to show it" do
-        threads = []
-        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") }
-        end
-
-        sleep(1.1) # Allow time for stats to update
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_idle cl_cancel_req sv_idle sv_used sv_tested sv_login maxwait].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-
-        expect(results["cl_waiting"]).to eq("2")
-        expect(results["cl_active"]).to eq("2")
-        expect(results["sv_active"]).to eq("2")
-
-        sleep(2.5) # Allow time for stats to update
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login].each do |s|
-          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
-        end
-        expect(results["cl_idle"]).to eq("4")
-        expect(results["sv_idle"]).to eq("2")
-
-        threads.map(&:join)
-        connections.map(&:close)
-      end
-
-      it "show correct max_wait" do
-        threads = []
-        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") }
-        end
-
-        sleep(2.5) # Allow time for stats to update
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-
-        expect(results["maxwait"]).to eq("1")
-        expect(results["maxwait_us"].to_i).to be_within(200_000).of(500_000)
-
-        sleep(4.5) # Allow time for stats to update
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        expect(results["maxwait"]).to eq("0")
-
-        threads.map(&:join)
-        connections.map(&:close)
-      end
-    end
-  end
-
-  describe "SHOW CLIENTS" do
-    it "reports correct number and application names" do
-      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
-      connections = Array.new(20) { |i| PG::connect("#{conn_str}?application_name=app#{i % 5}") }
-
-      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-      sleep(1) # Wait for stats to be updated
-
-      results = admin_conn.async_exec("SHOW CLIENTS")
-      expect(results.count).to eq(21) # count admin clients
-      expect(results.select { |c| c["application_name"] == "app3" ||  c["application_name"] == "app4" }.count).to eq(8)
-      expect(results.select { |c| c["database"] == "pgcat" }.count).to eq(1)
-
-      connections[0..5].map(&:close)
-      sleep(1) # Wait for stats to be updated
-      results = admin_conn.async_exec("SHOW CLIENTS")
-      expect(results.count).to eq(15)
-
-      connections[6..].map(&:close)
-      sleep(1) # Wait for stats to be updated
-      expect(admin_conn.async_exec("SHOW CLIENTS").count).to eq(1)
-      admin_conn.close
-    end
-
-    it "reports correct number of queries and transactions" do
-      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
-
-      connections = Array.new(2) { |i| PG::connect("#{conn_str}?application_name=app#{i}") }
-      connections.each do |c|
-        c.async_exec("SELECT 1")
-        c.async_exec("SELECT 2")
-        c.async_exec("SELECT 3")
-        c.async_exec("BEGIN")
-        c.async_exec("SELECT 4")
-        c.async_exec("SELECT 5")
-        c.async_exec("COMMIT")
-      end
-
-      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-      sleep(1) # Wait for stats to be updated
-
-      results = admin_conn.async_exec("SHOW CLIENTS")
-      expect(results.count).to eq(3)
-      normal_client_results = results.reject { |r| r["database"] == "pgcat" }
-      expect(normal_client_results[0]["transaction_count"]).to eq("4")
-      expect(normal_client_results[1]["transaction_count"]).to eq("4")
-      expect(normal_client_results[0]["query_count"]).to eq("7")
-      expect(normal_client_results[1]["query_count"]).to eq("7")
-
-      admin_conn.close
-      connections.map(&:close)
-    end
-  end
-
   describe "Manual Banning" do
     let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 10) }
     before do
@@ -401,7 +81,7 @@ describe "Admin" do
     end
   end
 
-  describe "SHOW users" do
+  describe "SHOW USERS" do
     it "returns the right users" do
       admin_conn = PG::connect(processes.pgcat.admin_connection_string)
       results = admin_conn.async_exec("SHOW USERS")[0]
@@ -410,4 +90,28 @@ describe "Admin" do
       expect(results["pool_mode"]).to eq("transaction")
     end
   end
+
+  describe "PAUSE" do
+    it "pauses all pools" do
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      results = admin_conn.async_exec("SHOW DATABASES").to_a
+      expect(results.map{ |r| r["paused"] }.uniq).to eq(["0"])
+
+      admin_conn.async_exec("PAUSE")
+
+      results = admin_conn.async_exec("SHOW DATABASES").to_a
+      expect(results.map{ |r| r["paused"] }.uniq).to eq(["1"])
+
+      admin_conn.async_exec("RESUME")
+
+      results = admin_conn.async_exec("SHOW DATABASES").to_a
+      expect(results.map{ |r| r["paused"] }.uniq).to eq(["0"])
+    end
+
+    it "handles errors" do
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      expect { admin_conn.async_exec("PAUSE foo").to_a }.to raise_error(PG::SystemError)
+      expect { admin_conn.async_exec("PAUSE foo,bar").to_a }.to raise_error(PG::SystemError)
+    end
+  end
 end

tests/ruby/auth_query_spec.rb

@@ -185,7 +185,7 @@ describe "Auth Query" do
               },
             }
           }
-        } 
+        }
 
         context 'and with cleartext passwords set' do
           it 'it uses local passwords' do

tests/ruby/copy_spec.rb

@@ -0,0 +1,102 @@
+# frozen_string_literal: true
+require_relative 'spec_helper'
+
+
+describe "COPY Handling" do
+  let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 5) }
+  before do
+    new_configs = processes.pgcat.current_config
+
+    # Allow connections in the pool to expire faster
+    new_configs["general"]["idle_timeout"] = 5
+    processes.pgcat.update_config(new_configs)
+    # We need to kill the old process that was using the default configs
+    processes.pgcat.stop
+    processes.pgcat.start
+    processes.pgcat.wait_until_ready
+  end
+
+  before do
+    processes.all_databases.first.with_connection do |conn|
+      conn.async_exec "CREATE TABLE copy_test_table (a TEXT,b TEXT,c TEXT,d TEXT)"
+    end
+  end
+
+  after do
+    processes.all_databases.first.with_connection do |conn|
+      conn.async_exec "DROP TABLE copy_test_table;"
+    end
+  end
+
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  describe "COPY FROM" do
+    context "within transaction" do
+      it "finishes within alloted time" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        Timeout.timeout(3) do
+          conn.async_exec("BEGIN")
+          conn.copy_data "COPY copy_test_table FROM STDIN CSV" do
+            sleep 0.5
+            conn.put_copy_data "some,data,to,copy\n"
+            conn.put_copy_data "more,data,to,copy\n"
+          end
+          conn.async_exec("COMMIT")
+        end
+
+        res = conn.async_exec("SELECT * FROM copy_test_table").to_a
+        expect(res).to eq([
+          {"a"=>"some", "b"=>"data", "c"=>"to", "d"=>"copy"},
+          {"a"=>"more", "b"=>"data", "c"=>"to", "d"=>"copy"}
+        ])
+      end
+    end
+
+    context "outside transaction" do
+      it "finishes within alloted time" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        Timeout.timeout(3) do
+          conn.copy_data "COPY copy_test_table FROM STDIN CSV" do
+            sleep 0.5
+            conn.put_copy_data "some,data,to,copy\n"
+            conn.put_copy_data "more,data,to,copy\n"
+          end
+        end
+
+        res = conn.async_exec("SELECT * FROM copy_test_table").to_a
+        expect(res).to eq([
+          {"a"=>"some", "b"=>"data", "c"=>"to", "d"=>"copy"},
+          {"a"=>"more", "b"=>"data", "c"=>"to", "d"=>"copy"}
+        ])
+      end
+    end
+  end
+
+  describe "COPY TO" do
+    before do
+      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      conn.async_exec("BEGIN")
+      conn.copy_data "COPY copy_test_table FROM STDIN CSV" do
+        conn.put_copy_data "some,data,to,copy\n"
+        conn.put_copy_data "more,data,to,copy\n"
+      end
+      conn.async_exec("COMMIT")
+      conn.close
+    end
+
+    it "works" do
+      res = []
+      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      conn.copy_data "COPY copy_test_table TO STDOUT CSV" do
+        while row=conn.get_copy_data
+          res << row
+        end
+      end
+      expect(res).to eq(["some,data,to,copy\n", "more,data,to,copy\n"])
+    end
+  end
+
+end

tests/ruby/helpers/auth_query_helper.rb

@@ -33,18 +33,18 @@ module Helpers
             "0" => {
               "database" => "shard0",
               "servers" => [
-                ["localhost", primary.port.to_s, "primary"],
-                ["localhost", replica.port.to_s, "replica"],
+                ["localhost", primary.port.to_i, "primary"],
+                ["localhost", replica.port.to_i, "replica"],
               ]
             },
           },
           "users" => { "0" => user.merge(config_user) }
         }
       }
-      pgcat_cfg["general"]["port"] = pgcat.port
+      pgcat_cfg["general"]["port"] = pgcat.port.to_i
       pgcat.update_config(pgcat_cfg)
       pgcat.start
-      
+
       pgcat.wait_until_ready(
         pgcat.connection_string(
           "sharded_db",
@@ -92,13 +92,13 @@ module Helpers
             "0" => {
               "database" => database,
               "servers" => [
-                ["localhost", primary.port.to_s, "primary"],
-                ["localhost", replica.port.to_s, "replica"],
+                ["localhost", primary.port.to_i, "primary"],
+                ["localhost", replica.port.to_i, "replica"],
               ]
             },
           },
           "users" => { "0" => user.merge(config_user) }
-        }                                  
+        }
       end
       # Main proxy configs
       pgcat_cfg["pools"] = {
@@ -109,7 +109,7 @@ module Helpers
       pgcat_cfg["general"]["port"] = pgcat.port
       pgcat.update_config(pgcat_cfg.deep_merge(extra_conf))
       pgcat.start
-      
+
       pgcat.wait_until_ready(pgcat.connection_string("sharded_db0", pg_user['username'], pg_user['password']))
 
       OpenStruct.new.tap do |struct|

tests/ruby/helpers/pg_instance.rb

@@ -7,10 +7,24 @@ class PgInstance
   attr_reader :password
   attr_reader :database_name
 
+  def self.mass_takedown(databases)
+    raise StandardError "block missing" unless block_given?
+
+    databases.each do |database|
+      database.toxiproxy.toxic(:limit_data, bytes: 1).toxics.each(&:save)
+    end
+    sleep 0.1
+    yield
+  ensure
+    databases.each do |database|
+      database.toxiproxy.toxics.each(&:destroy)
+    end
+  end
+
   def initialize(port, username, password, database_name)
-    @original_port = port
+    @original_port = port.to_i
     @toxiproxy_port = 10000 + port.to_i
-    @port = @toxiproxy_port
+    @port = @toxiproxy_port.to_i
 
     @username = username
     @password = password
@@ -48,9 +62,9 @@ class PgInstance
 
   def take_down
     if block_given?
-      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).apply { yield }
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 1).apply { yield }
     else
-      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).toxics.each(&:save)
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 1).toxics.each(&:save)
     end
   end
 
@@ -89,6 +103,6 @@ class PgInstance
   end
 
   def count_select_1_plus_2
-    with_connection { |c| c.async_exec("SELECT SUM(calls) FROM pg_stat_statements WHERE query = 'SELECT $1 + $2'")[0]["sum"].to_i }
+    with_connection { |c| c.async_exec("SELECT SUM(calls) FROM pg_stat_statements WHERE query LIKE '%SELECT $1 + $2%'")[0]["sum"].to_i }
   end
 end

tests/ruby/helpers/pgcat_helper.rb

@@ -34,12 +34,13 @@ module Helpers
           "load_balancing_mode" => lb_mode,
           "primary_reads_enabled" => true,
           "query_parser_enabled" => true,
+          "query_parser_read_write_splitting" => true,
           "automatic_sharding_key" => "data.id",
           "sharding_function" => "pg_bigint_hash",
           "shards" => {
-            "0" => { "database" => "shard0", "servers" => [["localhost", primary0.port.to_s, "primary"]] },
-            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_s, "primary"]] },
-            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_s, "primary"]] },
+            "0" => { "database" => "shard0", "servers" => [["localhost", primary0.port.to_i, "primary"]] },
+            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_i, "primary"]] },
+            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_i, "primary"]] },
           },
           "users" => { "0" => user },
           "plugins" => {
@@ -99,7 +100,7 @@ module Helpers
             "0" => {
               "database" => "shard0",
               "servers" => [
-                ["localhost", primary.port.to_s, "primary"]
+                ["localhost", primary.port.to_i, "primary"]
               ]
             },
           },
@@ -118,7 +119,7 @@ module Helpers
       end
     end
 
-    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info")
+    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info", pool_settings={})
       user = {
         "password" => "sharding_user",
         "pool_size" => pool_size,
@@ -134,28 +135,32 @@ module Helpers
       replica1 = PgInstance.new(8432, user["username"], user["password"], "shard0")
       replica2 = PgInstance.new(9432, user["username"], user["password"], "shard0")
 
+      pool_config = {
+        "default_role" => "any",
+        "pool_mode" => pool_mode,
+        "load_balancing_mode" => lb_mode,
+        "primary_reads_enabled" => false,
+        "query_parser_enabled" => false,
+        "sharding_function" => "pg_bigint_hash",
+        "shards" => {
+          "0" => {
+            "database" => "shard0",
+            "servers" => [
+              ["localhost", primary.port.to_i, "primary"],
+              ["localhost", replica0.port.to_i, "replica"],
+              ["localhost", replica1.port.to_i, "replica"],
+              ["localhost", replica2.port.to_i, "replica"]
+            ]
+          },
+        },
+        "users" => { "0" => user }
+      }
+
+      pool_config = pool_config.merge(pool_settings)
+
       # Main proxy configs
       pgcat_cfg["pools"] = {
-        "#{pool_name}" => {
-          "default_role" => "any",
-          "pool_mode" => pool_mode,
-          "load_balancing_mode" => lb_mode,
-          "primary_reads_enabled" => false,
-          "query_parser_enabled" => false,
-          "sharding_function" => "pg_bigint_hash",
-          "shards" => {
-            "0" => {
-              "database" => "shard0",
-              "servers" => [
-                ["localhost", primary.port.to_s, "primary"],
-                ["localhost", replica0.port.to_s, "replica"],
-                ["localhost", replica1.port.to_s, "replica"],
-                ["localhost", replica2.port.to_s, "replica"]
-              ]
-            },
-          },
-          "users" => { "0" => user }
-        }
+        "#{pool_name}" => pool_config,
       }
       pgcat_cfg["general"]["port"] = pgcat.port
       pgcat.update_config(pgcat_cfg)

tests/ruby/helpers/pgcat_process.rb

@@ -1,8 +1,10 @@
 require 'pg'
-require 'toml'
+require 'json'
+require 'tempfile'
 require 'fileutils'
 require 'securerandom'
 
+class ConfigReloadFailed < StandardError; end
 class PgcatProcess
   attr_reader :port
   attr_reader :pid
@@ -18,7 +20,7 @@ class PgcatProcess
   end
 
   def initialize(log_level)
-    @env = {"RUST_LOG" => log_level}
+    @env = {}
     @port = rand(20000..32760)
     @log_level = log_level
     @log_filename = "/tmp/pgcat_log_#{SecureRandom.urlsafe_base64}.log"
@@ -30,7 +32,7 @@ class PgcatProcess
                      '../../target/debug/pgcat'
                    end
 
-    @command = "#{command_path} #{@config_filename}"
+    @command = "#{command_path} #{@config_filename} --log-level #{@log_level}"
 
     FileUtils.cp("../../pgcat.toml", @config_filename)
     cfg = current_config
@@ -46,22 +48,34 @@ class PgcatProcess
 
   def update_config(config_hash)
     @original_config = current_config
-    output_to_write = TOML::Generator.new(config_hash).body
-    output_to_write = output_to_write.gsub(/,\s*["|'](\d+)["|']\s*,/, ',\1,')
-    output_to_write = output_to_write.gsub(/,\s*["|'](\d+)["|']\s*\]/, ',\1]')
-    File.write(@config_filename, output_to_write)
+    Tempfile.create('json_out', '/tmp') do |f|
+      f.write(config_hash.to_json)
+      f.flush
+      `cat #{f.path} | yj -jt > #{@config_filename}`
+    end
   end
 
   def current_config
-    loadable_string = File.read(@config_filename)
-    loadable_string = loadable_string.gsub(/,\s*(\d+)\s*,/,  ', "\1",')
-    loadable_string = loadable_string.gsub(/,\s*(\d+)\s*\]/, ', "\1"]')
-    TOML.load(loadable_string)
+    JSON.parse(`cat #{@config_filename} | yj -tj`)
+  end
+
+  def raw_config_file
+    File.read(@config_filename)
   end
 
   def reload_config
-    `kill -s HUP #{@pid}`
-    sleep 0.5
+    conn = PG.connect(admin_connection_string)
+
+    conn.async_exec("RELOAD")
+  rescue PG::ConnectionBad => e
+    errors = logs.split("Reloading config").last
+    errors = errors.gsub(/\e\[([;\d]+)?m/, '') # Remove color codes
+    errors = errors.
+                split("\n").select{|line| line.include?("ERROR") }.
+                map { |line| line.split("pgcat::config: ").last }
+    raise ConfigReloadFailed, errors.join("\n")
+  ensure
+    conn&.close
   end
 
   def start
@@ -112,10 +126,16 @@ class PgcatProcess
     "postgresql://#{username}:#{password}@0.0.0.0:#{@port}/pgcat"
   end
 
-  def connection_string(pool_name, username, password = nil)
+  def connection_string(pool_name, username, password = nil, parameters: {})
     cfg = current_config
     user_idx, user_obj = cfg["pools"][pool_name]["users"].detect { |k, user| user["username"] == username }
-    "postgresql://#{username}:#{password || user_obj["password"]}@0.0.0.0:#{@port}/#{pool_name}"
+    connection_string = "postgresql://#{username}:#{password || user_obj["password"]}@0.0.0.0:#{@port}/#{pool_name}"
+
+    # Add the additional parameters to the connection string
+    parameter_string = parameters.map { |key, value| "#{key}=#{value}" }.join("&")
+    connection_string += "?#{parameter_string}" unless parameter_string.empty?
+
+    connection_string
   end
 
   def example_connection_string

tests/ruby/mirrors_spec.rb

@@ -11,9 +11,9 @@ describe "Query Mirroing" do
   before do
     new_configs = processes.pgcat.current_config
     new_configs["pools"]["sharded_db"]["shards"]["0"]["mirrors"] = [
-      [mirror_host, mirror_pg.port.to_s, "0"],
-      [mirror_host, mirror_pg.port.to_s, "0"],
-      [mirror_host, mirror_pg.port.to_s, "0"],
+      [mirror_host, mirror_pg.port.to_i, 0],
+      [mirror_host, mirror_pg.port.to_i, 0],
+      [mirror_host, mirror_pg.port.to_i, 0],
     ]
     processes.pgcat.update_config(new_configs)
     processes.pgcat.reload_config
@@ -31,7 +31,8 @@ describe "Query Mirroing" do
     runs.times { conn.async_exec("SELECT 1 + 2") }
     sleep 0.5
     expect(processes.all_databases.first.count_select_1_plus_2).to eq(runs)
-    expect(mirror_pg.count_select_1_plus_2).to eq(runs * 3)
+    # Allow some slack in mirroring successes
+    expect(mirror_pg.count_select_1_plus_2).to be > ((runs - 5) * 3)
   end
 
   context "when main server connection is closed" do
@@ -42,9 +43,9 @@ describe "Query Mirroing" do
         new_configs = processes.pgcat.current_config
         new_configs["pools"]["sharded_db"]["idle_timeout"] = 5000 + i
         new_configs["pools"]["sharded_db"]["shards"]["0"]["mirrors"] = [
-          [mirror_host, mirror_pg.port.to_s, "0"],
-          [mirror_host, mirror_pg.port.to_s, "0"],
-          [mirror_host, mirror_pg.port.to_s, "0"],
+          [mirror_host, mirror_pg.port.to_i, 0],
+          [mirror_host, mirror_pg.port.to_i, 0],
+          [mirror_host, mirror_pg.port.to_i, 0],
         ]
         processes.pgcat.update_config(new_configs)
         processes.pgcat.reload_config

tests/ruby/misc_spec.rb

@@ -221,7 +221,7 @@ describe "Miscellaneous" do
         conn.close
       end
 
-      it "Does not send DISCARD ALL unless necessary" do
+      it "Does not send RESET ALL unless necessary" do
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
           conn.async_exec("SET SERVER ROLE to 'primary'")
@@ -229,7 +229,7 @@ describe "Miscellaneous" do
           conn.close
         end
 
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
 
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
@@ -239,7 +239,7 @@ describe "Miscellaneous" do
           conn.close
         end
 
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
+        expect(processes.primary.count_query("RESET ALL")).to eq(10)
       end
 
       it "Resets server roles correctly" do
@@ -252,7 +252,7 @@ describe "Miscellaneous" do
         end
 
         expect(processes.primary.count_query("RESET ROLE")).to eq(10)
-      end 
+      end
     end
 
     context "transaction mode" do
@@ -273,7 +273,7 @@ describe "Miscellaneous" do
         end
       end
 
-      it "Does not send DISCARD ALL unless necessary" do
+      it "Does not send RESET ALL unless necessary" do
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
           conn.async_exec("SET SERVER ROLE to 'primary'")
@@ -282,7 +282,7 @@ describe "Miscellaneous" do
           conn.close
         end
 
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
 
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
@@ -292,8 +292,32 @@ describe "Miscellaneous" do
           conn.close
         end
 
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
+        expect(processes.primary.count_query("RESET ALL")).to eq(10)
       end
+
+      it "Respects tracked parameters on startup" do
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user", parameters: { "application_name" => "my_pgcat_test" }))
+
+        expect(conn.async_exec("SHOW application_name")[0]["application_name"]).to eq("my_pgcat_test")
+        conn.close
+      end
+
+      it "Respect tracked parameter on set statemet" do
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+
+        conn.async_exec("SET application_name to 'my_pgcat_test'")
+        expect(conn.async_exec("SHOW application_name")[0]["application_name"]).to eq("my_pgcat_test")
+      end
+
+
+      it "Ignore untracked parameter on set statemet" do
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        orignal_statement_timeout = conn.async_exec("SHOW statement_timeout")[0]["statement_timeout"]
+
+        conn.async_exec("SET statement_timeout to 1500")
+        expect(conn.async_exec("SHOW statement_timeout")[0]["statement_timeout"]).to eq(orignal_statement_timeout)
+      end
+
     end
 
     context "transaction mode with transactions" do
@@ -307,7 +331,7 @@ describe "Miscellaneous" do
           conn.async_exec("COMMIT")
           conn.close
         end
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
 
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
@@ -317,7 +341,30 @@ describe "Miscellaneous" do
           conn.async_exec("COMMIT")
           conn.close
         end
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+      end
+    end
+
+    context "server cleanup disabled" do
+      let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 1, "transaction", "random", "info", { "cleanup_server_connections" => false }) }
+
+      it "will not clean up connection state" do
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        processes.primary.reset_stats
+        conn.async_exec("SET statement_timeout TO 1000")
+        conn.close
+
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+      end
+
+      it "will not clean up prepared statements" do
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        processes.primary.reset_stats
+        conn.async_exec("PREPARE prepared_q (int) AS SELECT $1")
+
+        conn.close
+
+        expect(processes.primary.count_query("RESET ALL")).to eq(0)
       end
     end
   end
@@ -327,10 +374,9 @@ describe "Miscellaneous" do
       before do
         current_configs = processes.pgcat.current_config
         correct_idle_client_transaction_timeout = current_configs["general"]["idle_client_in_transaction_timeout"]
-        puts(current_configs["general"]["idle_client_in_transaction_timeout"])
-  
+
         current_configs["general"]["idle_client_in_transaction_timeout"] = 0
-  
+
         processes.pgcat.update_config(current_configs) # with timeout 0
         processes.pgcat.reload_config
       end
@@ -348,9 +394,9 @@ describe "Miscellaneous" do
     context "idle transaction timeout set to 500ms" do
       before do
         current_configs = processes.pgcat.current_config
-        correct_idle_client_transaction_timeout = current_configs["general"]["idle_client_in_transaction_timeout"]  
+        correct_idle_client_transaction_timeout = current_configs["general"]["idle_client_in_transaction_timeout"]
         current_configs["general"]["idle_client_in_transaction_timeout"] = 500
-  
+
         processes.pgcat.update_config(current_configs) # with timeout 500
         processes.pgcat.reload_config
       end
@@ -369,7 +415,7 @@ describe "Miscellaneous" do
         conn.async_exec("BEGIN")
         conn.async_exec("SELECT 1")
         sleep(1) # above 500ms
-        expect{ conn.async_exec("COMMIT") }.to raise_error(PG::SystemError, /idle transaction timeout/) 
+        expect{ conn.async_exec("COMMIT") }.to raise_error(PG::SystemError, /idle transaction timeout/)
         conn.async_exec("SELECT 1") # should be able to send another query
         conn.close
       end

tests/ruby/prepared_spec.rb

@@ -0,0 +1,214 @@
+require_relative 'spec_helper'
+
+describe 'Prepared statements' do
+  let(:pool_size) { 5 }
+  let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", pool_size) }
+  let(:prepared_statements_cache_size) { 100 }
+  let(:server_round_robin) { false }
+
+  before do
+    new_configs = processes.pgcat.current_config
+    new_configs["general"]["server_round_robin"] = server_round_robin
+    new_configs["pools"]["sharded_db"]["prepared_statements_cache_size"] = prepared_statements_cache_size
+    new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] =  pool_size
+    processes.pgcat.update_config(new_configs)
+    processes.pgcat.reload_config
+  end
+
+  context 'when trying prepared statements' do
+    it 'it allows unparameterized statements to succeed' do
+      conn1 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+      conn2 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      prepared_query = "SELECT 1"
+      
+      # prepare query on server 1 and client 1
+      conn1.prepare('statement1', prepared_query)
+      conn1.exec_prepared('statement1')
+      
+      conn2.transaction do
+        # Claim server 1 with client 2
+        conn2.exec("SELECT 2")
+      
+        # Client 1 now runs the prepared query, and it's automatically
+        # prepared on server 2
+        conn1.prepare('statement2', prepared_query)
+        conn1.exec_prepared('statement2')
+      
+        # Client 2 now prepares the same query that was already
+        # prepared on server 1. And PgBouncer reuses that already
+        # prepared query for this different client.
+        conn2.prepare('statement3', prepared_query)
+        conn2.exec_prepared('statement3')
+        end
+    ensure
+        conn1.close if conn1
+        conn2.close if conn2
+    end
+
+    it 'it allows parameterized statements to succeed' do
+      conn1 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+      conn2 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      prepared_query = "SELECT $1"
+      
+      # prepare query on server 1 and client 1
+      conn1.prepare('statement1', prepared_query)
+      conn1.exec_prepared('statement1', [1])
+    
+      conn2.transaction do
+        # Claim server 1 with client 2
+        conn2.exec("SELECT 2")
+      
+        # Client 1 now runs the prepared query, and it's automatically
+        # prepared on server 2
+        conn1.prepare('statement2', prepared_query)
+        conn1.exec_prepared('statement2', [1])
+      
+        # Client 2 now prepares the same query that was already
+        # prepared on server 1. And PgBouncer reuses that already
+        # prepared query for this different client.
+        conn2.prepare('statement3', prepared_query)
+        conn2.exec_prepared('statement3', [1])
+      end
+    ensure
+       conn1.close if conn1
+       conn2.close if conn2
+
+    end
+  end
+
+  context 'when trying large packets' do
+    it "works with large parse" do
+      conn1 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      long_string = "1" * 4096 * 10
+      prepared_query = "SELECT '#{long_string}'"
+    
+  
+      # prepare query on server 1 and client 1
+      conn1.prepare('statement1', prepared_query)
+      result = conn1.exec_prepared('statement1')
+    
+      # assert result matches long_string
+      expect(result.getvalue(0, 0)).to eq(long_string)
+    ensure
+      conn1.close if conn1
+    end
+
+    it "works with large bind" do
+      conn1 = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+    
+      long_string = "1" * 4096 * 10
+      prepared_query = "SELECT $1::text"
+    
+      # prepare query on server 1 and client 1
+      conn1.prepare('statement1', prepared_query)
+      result = conn1.exec_prepared('statement1', [long_string])
+    
+      # assert result matches long_string
+      expect(result.getvalue(0, 0)).to eq(long_string)
+    ensure
+      conn1.close if conn1
+    end
+  end
+
+  context 'when statement cache is smaller than set of unqiue statements' do
+    let(:prepared_statements_cache_size) { 1 }
+    let(:pool_size) { 1 }
+
+    it "evicts all but 1 statement from the server cache" do
+      conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      5.times do |i|
+        prepared_query = "SELECT '#{i}'"
+        conn.prepare("statement#{i}", prepared_query)
+        result = conn.exec_prepared("statement#{i}")
+        expect(result.getvalue(0, 0)).to eq(i.to_s)
+      end
+
+      # Check number of prepared statements (expected: 1)
+      n_statements = conn.exec("SELECT count(*) FROM pg_prepared_statements").getvalue(0, 0).to_i
+      expect(n_statements).to eq(1)
+    end
+  end
+
+  context 'when statement cache is larger than set of unqiue statements' do
+    let(:pool_size) { 1 }
+
+    it "does not evict any of the statements from the cache" do
+      # cache size 5
+      conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      5.times do |i|
+        prepared_query = "SELECT '#{i}'"
+        conn.prepare("statement#{i}", prepared_query)
+        result = conn.exec_prepared("statement#{i}")
+        expect(result.getvalue(0, 0)).to eq(i.to_s)
+      end
+
+      # Check number of prepared statements (expected: 1)
+      n_statements = conn.exec("SELECT count(*) FROM pg_prepared_statements").getvalue(0, 0).to_i
+      expect(n_statements).to eq(5)
+    end
+  end
+
+  context 'when preparing the same query' do
+    let(:prepared_statements_cache_size) { 5 }
+    let(:pool_size) { 5 }
+
+    it "reuses statement cache when there are different statement names on the same connection" do
+      conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      10.times do |i|
+        statement_name = "statement_#{i}"
+        conn.prepare(statement_name, 'SELECT $1::int')
+        conn.exec_prepared(statement_name, [1])
+      end
+
+      # Check number of prepared statements (expected: 1)
+      n_statements = conn.exec("SELECT count(*) FROM pg_prepared_statements").getvalue(0, 0).to_i
+      expect(n_statements).to eq(1)
+    end
+
+    it "reuses statement cache when there are different statement names on different connections" do
+      10.times do |i|
+        conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+        statement_name = "statement_#{i}"
+        conn.prepare(statement_name, 'SELECT $1::int')
+        conn.exec_prepared(statement_name, [1])
+      end
+
+      # Check number of prepared statements (expected: 1)
+      conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+      n_statements = conn.exec("SELECT count(*) FROM pg_prepared_statements").getvalue(0, 0).to_i
+      expect(n_statements).to eq(1)
+    end
+  end
+
+  context 'when reloading config' do
+    let(:pool_size) { 1 }
+
+    it "test_reload_config" do
+      conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+
+      # prepare query
+      conn.prepare('statement1', 'SELECT 1')
+      conn.exec_prepared('statement1')
+
+      # Reload config which triggers pool recreation
+      new_configs = processes.pgcat.current_config
+      new_configs["pools"]["sharded_db"]["prepared_statements_cache_size"] = prepared_statements_cache_size + 1
+      processes.pgcat.update_config(new_configs)
+      processes.pgcat.reload_config
+
+      # check that we're starting with no prepared statements on the server
+      conn_check = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
+      n_statements = conn_check.exec("SELECT count(*) FROM pg_prepared_statements").getvalue(0, 0).to_i
+      expect(n_statements).to eq(0)
+
+      # still able to run prepared query
+      conn.exec_prepared('statement1')
+    end
+  end
+end

tests/ruby/sharding_spec.rb

@@ -7,11 +7,11 @@ describe "Sharding" do
 
   before do
     conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
-
     # Setup the sharding data
     3.times do |i|
       conn.exec("SET SHARD TO '#{i}'")
-      conn.exec("DELETE FROM data WHERE id > 0")
+
+      conn.exec("DELETE FROM data WHERE id > 0") rescue nil
     end
 
     18.times do |i|
@@ -19,10 +19,11 @@ describe "Sharding" do
       conn.exec("SET SHARDING KEY TO '#{i}'")
       conn.exec("INSERT INTO data (id, value) VALUES (#{i}, 'value_#{i}')")
     end
+
+    conn.close
   end
 
   after do
-
     processes.all_databases.map(&:reset)
     processes.pgcat.shutdown
   end
@@ -48,4 +49,148 @@ describe "Sharding" do
       end
     end
   end
+
+  describe "no_shard_specified_behavior config" do
+    context "when default shard number is invalid" do
+      it "prevents config reload" do
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+
+        current_configs = processes.pgcat.current_config
+        current_configs["pools"]["sharded_db"]["default_shard"] = "shard_99"
+
+        processes.pgcat.update_config(current_configs)
+
+        expect { processes.pgcat.reload_config }.to raise_error(ConfigReloadFailed, /Invalid shard 99/)
+      end
+    end
+  end
+
+  describe "comment-based routing" do
+    context "when no configs are set" do
+      it "routes queries with a shard_id comment to the default shard" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        10.times { conn.async_exec("/* shard_id: 2 */ SELECT 1 + 2") }
+
+        expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([10, 0, 0])
+      end
+
+      it "does not honor no_shard_specified_behavior directives" do
+      end
+    end
+
+    [
+      ["shard_id_regex", "/\\* the_shard_id: (\\d+) \\*/", "/* the_shard_id: 1 */"],
+      ["sharding_key_regex", "/\\* the_sharding_key: (\\d+) \\*/", "/* the_sharding_key: 3 */"],
+    ].each do |config_name, config_value, comment_to_use|
+      context "when #{config_name} config is set" do
+        let(:no_shard_specified_behavior) { nil }
+
+        before do
+          admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+
+          current_configs = processes.pgcat.current_config
+          current_configs["pools"]["sharded_db"][config_name] = config_value
+          if no_shard_specified_behavior
+            current_configs["pools"]["sharded_db"]["default_shard"] = no_shard_specified_behavior
+          else
+            current_configs["pools"]["sharded_db"].delete("default_shard")
+          end
+
+          processes.pgcat.update_config(current_configs)
+          processes.pgcat.reload_config
+        end
+
+        it "routes queries with a shard_id comment to the correct shard" do
+          conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          25.times { conn.async_exec("#{comment_to_use} SELECT 1 + 2") }
+
+          expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([0, 25, 0])
+        end
+
+        context "when no_shard_specified_behavior config is set to random" do
+          let(:no_shard_specified_behavior) { "random" }
+
+          context "with no shard comment" do
+            it "sends queries to random shard" do
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              25.times { conn.async_exec("SELECT 1 + 2") }
+
+              expect(processes.all_databases.map(&:count_select_1_plus_2).all?(&:positive?)).to be true
+            end
+          end
+
+          context "with a shard comment" do
+            it "honors the comment" do
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              25.times { conn.async_exec("#{comment_to_use} SELECT 1 + 2") }
+
+              expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([0, 25, 0])
+            end
+          end
+        end
+
+        context "when no_shard_specified_behavior config is set to random_healthy" do
+          let(:no_shard_specified_behavior) { "random_healthy" }
+
+          context "with no shard comment" do
+            it "sends queries to random healthy shard" do
+
+              good_databases = [processes.all_databases[0], processes.all_databases[2]]
+              bad_database = processes.all_databases[1]
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              250.times { conn.async_exec("SELECT 99") }
+              bad_database.take_down do
+                250.times do
+                  conn.async_exec("SELECT 99")
+                rescue PG::ConnectionBad => e
+                  conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+                end
+              end
+
+              # Routes traffic away from bad shard
+              25.times { conn.async_exec("SELECT 1 + 2") }
+              expect(good_databases.map(&:count_select_1_plus_2).all?(&:positive?)).to be true
+              expect(bad_database.count_select_1_plus_2).to eq(0)
+
+              # Routes traffic to the bad shard if the shard_id is specified
+              25.times { conn.async_exec("#{comment_to_use} SELECT 1 + 2") }
+              bad_database = processes.all_databases[1]
+              expect(bad_database.count_select_1_plus_2).to eq(25)
+            end
+          end
+
+          context "with a shard comment" do
+            it "honors the comment" do
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              25.times { conn.async_exec("#{comment_to_use} SELECT 1 + 2") }
+
+              expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([0, 25, 0])
+            end
+          end
+        end
+
+        context "when no_shard_specified_behavior config is set to shard_x" do
+          let(:no_shard_specified_behavior) { "shard_2" }
+
+          context "with no shard comment" do
+            it "sends queries to the specified shard" do
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              25.times { conn.async_exec("SELECT 1 + 2") }
+
+              expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([0, 0, 25])
+            end
+          end
+
+          context "with a shard comment" do
+            it "honors the comment" do
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              25.times { conn.async_exec("#{comment_to_use} SELECT 1 + 2") }
+
+              expect(processes.all_databases.map(&:count_select_1_plus_2)).to eq([0, 25, 0])
+            end
+          end
+        end
+      end
+    end
+  end
 end

tests/ruby/stats_spec.rb

@@ -0,0 +1,411 @@
+# frozen_string_literal: true
+require 'open3'
+require_relative 'spec_helper'
+
+describe "Stats" do
+  let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 10) }
+  let(:pgcat_conn_str) { processes.pgcat.connection_string("sharded_db", "sharding_user") }
+
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  describe "SHOW STATS" do
+    context "clients connect and make one query" do
+      it "updates *_query_time and *_wait_time" do
+        connections = Array.new(3) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new { c.async_exec("SELECT pg_sleep(0.25)") }
+        end
+        sleep(1)
+        connections.map(&:close)
+
+        # wait for averages to be calculated, we shouldn't do this too often
+        sleep(15.5)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW STATS")[0]
+        admin_conn.close
+        expect(results["total_query_time"].to_i).to be_within(200).of(750)
+        expect(results["avg_query_time"].to_i).to be_within(50).of(250)
+
+        expect(results["total_wait_time"].to_i).to_not eq(0)
+        expect(results["avg_wait_time"].to_i).to_not eq(0)
+      end
+    end
+  end
+
+  describe "SHOW POOLS" do
+    context "bad credentials" do
+      it "does not change any stats" do
+        bad_password_url = URI(pgcat_conn_str)
+        bad_password_url.password = "wrong"
+        expect { PG::connect("#{bad_password_url.to_s}?application_name=bad_password") }.to raise_error(PG::ConnectionBad)
+
+        sleep(1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["sv_idle"]).to eq("1")
+      end
+    end
+
+    context "bad database name" do
+      it "does not change any stats" do
+        bad_db_url = URI(pgcat_conn_str)
+        bad_db_url.path = "/wrong_db"
+        expect { PG::connect("#{bad_db_url.to_s}?application_name=bad_db") }.to raise_error(PG::ConnectionBad)
+
+        sleep(1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["sv_idle"]).to eq("1")
+      end
+    end
+
+    context "client connects but issues no queries" do
+      it "only affects cl_idle stats" do
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+
+        before_test = admin_conn.async_exec("SHOW POOLS")[0]["sv_idle"]
+        connections = Array.new(20) { PG::connect(pgcat_conn_str) }
+        sleep(1)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("20")
+        expect(results["sv_idle"]).to eq(before_test)
+
+        connections.map(&:close)
+        sleep(1.1)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_idle cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq(before_test)
+      end
+    end
+
+    context "clients connect and make one query" do
+      it "only affects cl_idle, sv_idle stats" do
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new { c.async_exec("SELECT pg_sleep(2.5)") }
+        end
+
+        sleep(1.1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_waiting cl_cancel_req sv_idle sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_active"]).to eq("5")
+        expect(results["sv_active"]).to eq("5")
+
+        sleep(3)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("5")
+        expect(results["sv_idle"]).to eq("5")
+
+        connections.map(&:close)
+        sleep(1)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq("5")
+      end
+    end
+
+    context "client connects and opens a transaction and closes connection uncleanly" do
+      it "produces correct statistics" do
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new do
+            c.async_exec("BEGIN")
+            c.async_exec("SELECT pg_sleep(0.01)")
+            c.close
+          end
+        end
+
+        sleep(1.1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq("5")
+      end
+    end
+
+    context "client fail to checkout connection from the pool" do
+      it "counts clients as idle" do
+        new_configs = processes.pgcat.current_config
+        new_configs["general"]["connect_timeout"] = 500
+        new_configs["general"]["ban_time"] = 1
+        new_configs["general"]["shutdown_timeout"] = 1
+        new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
+        processes.pgcat.update_config(new_configs)
+        processes.pgcat.reload_config
+
+        threads = []
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1)") rescue PG::SystemError }
+        end
+
+        sleep(2)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("5")
+        expect(results["sv_idle"]).to eq("1")
+
+        threads.map(&:join)
+        connections.map(&:close)
+      end
+    end
+
+    context "clients connects and disconnect normally" do
+      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
+
+      it 'shows the same number of clients before and after' do
+        clients_before = clients_connected_to_pool(processes: processes)
+        threads = []
+        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT 1") rescue nil }
+        end
+        clients_between = clients_connected_to_pool(processes: processes)
+        expect(clients_before).not_to eq(clients_between)
+        connections.each(&:close)
+        clients_after = clients_connected_to_pool(processes: processes)
+        expect(clients_before).to eq(clients_after)
+      end
+    end
+
+    context "clients connects and disconnect abruptly" do
+      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 10) }
+
+      it 'shows the same number of clients before and after' do
+        threads = []
+        connections = Array.new(2) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT 1") }
+        end
+        clients_before = clients_connected_to_pool(processes: processes)
+        random_string = (0...8).map { (65 + rand(26)).chr }.join
+        connection_string = "#{pgcat_conn_str}?application_name=#{random_string}"
+        faulty_client = Process.spawn("psql -Atx #{connection_string} >/dev/null")
+        sleep(1)
+        # psql starts two processes, we only know the pid of the parent, this
+        # ensure both are killed
+        `pkill -9 -f '#{random_string}'`
+        Process.wait(faulty_client)
+        clients_after = clients_connected_to_pool(processes: processes)
+        expect(clients_before).to eq(clients_after)
+      end
+    end
+
+    context "clients overwhelm server pools" do
+      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
+
+      it "cl_waiting is updated to show it" do
+        threads = []
+        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") }
+        end
+
+        sleep(1.1) # Allow time for stats to update
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_cancel_req sv_idle sv_used sv_tested sv_login].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["cl_waiting"]).to eq("2")
+        expect(results["cl_active"]).to eq("2")
+        expect(results["sv_active"]).to eq("2")
+
+        sleep(2.5) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("4")
+        expect(results["sv_idle"]).to eq("2")
+
+        threads.map(&:join)
+        connections.map(&:close)
+      end
+
+      it "show correct max_wait" do
+        threads = []
+        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") rescue nil }
+        end
+
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+
+        # two connections waiting => they report wait time
+        sleep(1.1) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        expect(results["maxwait"]).to eq("1")
+        expect(results["maxwait_us"].to_i).to be_within(200_000).of(100_000)
+
+        sleep(2.5) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+
+        # no connections waiting => no reported wait time
+        expect(results["maxwait"]).to eq("0")
+        expect(results["maxwait_us"]).to eq("0")
+        connections.map(&:close)
+
+        sleep(4.5) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        expect(results["maxwait"]).to eq("0")
+
+        threads.map(&:join)
+      end
+    end
+  end
+
+  describe "SHOW CLIENTS" do
+    it "reports correct number and application names" do
+      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
+      connections = Array.new(20) { |i| PG::connect("#{conn_str}?application_name=app#{i % 5}") }
+
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      sleep(1) # Wait for stats to be updated
+
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(21) # count admin clients
+      expect(results.select { |c| c["application_name"] == "app3" ||  c["application_name"] == "app4" }.count).to eq(8)
+      expect(results.select { |c| c["database"] == "pgcat" }.count).to eq(1)
+
+      connections[0..5].map(&:close)
+      sleep(1) # Wait for stats to be updated
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(15)
+
+      connections[6..].map(&:close)
+      sleep(1) # Wait for stats to be updated
+      expect(admin_conn.async_exec("SHOW CLIENTS").count).to eq(1)
+      admin_conn.close
+    end
+
+    it "reports correct number of queries and transactions" do
+      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
+
+      connections = Array.new(2) { |i| PG::connect("#{conn_str}?application_name=app#{i}") }
+      connections.each do |c|
+        c.async_exec("SELECT 1")
+        c.async_exec("SELECT 2")
+        c.async_exec("SELECT 3")
+        c.async_exec("BEGIN")
+        c.async_exec("SELECT 4")
+        c.async_exec("SELECT 5")
+        c.async_exec("COMMIT")
+      end
+
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      sleep(1) # Wait for stats to be updated
+
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(3)
+      normal_client_results = results.reject { |r| r["database"] == "pgcat" }
+      expect(normal_client_results[0]["transaction_count"]).to eq("4")
+      expect(normal_client_results[1]["transaction_count"]).to eq("4")
+      expect(normal_client_results[0]["query_count"]).to eq("7")
+      expect(normal_client_results[1]["query_count"]).to eq("7")
+
+      admin_conn.close
+      connections.map(&:close)
+    end
+
+    context "when client has waited for a server" do
+      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
+
+      it "shows correct maxwait" do
+        threads = []
+        connections = Array.new(3) { |i| PG::connect("#{pgcat_conn_str}?application_name=app#{i}") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") rescue nil }
+        end
+
+        sleep(2.5) # Allow time for stats to update
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW CLIENTS")
+
+        normal_client_results = results.reject { |r| r["database"] == "pgcat" }
+
+        non_waiting_clients = normal_client_results.select { |c| c["maxwait"] == "0" }
+        waiting_clients = normal_client_results.select { |c| c["maxwait"].to_i > 0 }
+
+        expect(non_waiting_clients.count).to eq(2)
+        non_waiting_clients.each do |client|
+          expect(client["maxwait_us"].to_i).to be_between(0, 50_000)
+        end
+
+        expect(waiting_clients.count).to eq(1)
+        waiting_clients.each do |client|
+          expect(client["maxwait_us"].to_i).to be_within(200_000).of(500_000)
+        end
+
+        admin_conn.close
+        connections.map(&:close)
+      end
+    end
+  end
+
+
+  describe "Query Storm" do
+    context "when the proxy receives overwhelmingly large number of short quick queries" do
+      it "should not have lingering clients or active servers" do
+        new_configs = processes.pgcat.current_config
+
+        new_configs["general"]["connect_timeout"] = 500
+        new_configs["general"]["ban_time"] = 1
+        new_configs["general"]["shutdown_timeout"] = 1
+        new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
+        processes.pgcat.update_config(new_configs)
+        processes.pgcat.reload_config
+
+        Array.new(40) do
+          Thread.new do
+            conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+            conn.async_exec("SELECT pg_sleep(0.1)")
+          rescue PG::SystemError
+          ensure
+            conn.close
+          end
+        end.each(&:join)
+
+        sleep 1
+
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_waiting cl_cancel_req sv_used sv_tested sv_login].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        admin_conn.close
+      end
+    end
+  end
+end

tests/rust/.gitignore

@@ -0,0 +1 @@
+target/

tests/rust/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "rust"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+sqlx = { version = "0.6.2", features = [ "runtime-tokio-rustls", "postgres", "json", "tls", "migrate", "time", "uuid", "ipnetwork"] }
+tokio = { version = "1", features = ["full"] }

tests/rust/src/main.rs

@@ -0,0 +1,36 @@
+#[tokio::main]
+async fn main() {
+    test_prepared_statements().await;
+}
+
+async fn test_prepared_statements() {
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(5)
+        .connect("postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db")
+        .await
+        .unwrap();
+
+    let mut handles = Vec::new();
+
+    for _ in 0..5 {
+        let pool = pool.clone();
+        let handle = tokio::task::spawn(async move {
+            for _ in 0..1000 {
+                match sqlx::query("SELECT one").fetch_all(&pool).await {
+                    Ok(_) => (),
+                    Err(err) => {
+                        if err.to_string().contains("prepared statement") {
+                            panic!("prepared statement error: {}", err);
+                        }
+                    }
+                }
+            }
+        });
+
+        handles.push(handle);
+    }
+
+    for handle in handles {
+        handle.await.unwrap();
+    }
+}

utilities/deb.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+#
+# Build an Ubuntu deb.
+#
+script_dir=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+deb_dir="/tmp/pgcat-build"
+export PACKAGE_VERSION=${1:-"1.1.1"}
+if [[ $(arch) == "x86_64" ]]; then
+  export ARCH=amd64
+else
+  export ARCH=arm64
+fi
+
+cd "$script_dir/.."
+cargo build --release
+
+rm -rf "$deb_dir"
+mkdir -p "$deb_dir/DEBIAN"
+mkdir -p "$deb_dir/usr/bin"
+mkdir -p "$deb_dir/etc/systemd/system"
+
+cp target/release/pgcat "$deb_dir/usr/bin/pgcat"
+chmod +x "$deb_dir/usr/bin/pgcat"
+
+cp pgcat.toml "$deb_dir/etc/pgcat.example.toml"
+cp pgcat.service "$deb_dir/etc/systemd/system/pgcat.service"
+
+(cat control | envsubst) > "$deb_dir/DEBIAN/control"
+cp postinst "$deb_dir/DEBIAN/postinst"
+cp postrm "$deb_dir/DEBIAN/postrm"
+cp prerm "$deb_dir/DEBIAN/prerm"
+
+chmod +x ${deb_dir}/DEBIAN/post*
+chmod +x ${deb_dir}/DEBIAN/pre*
+
+dpkg-deb \
+  --root-owner-group \
+  -z1 \
+  --build "$deb_dir" \
+  pgcat-${PACKAGE_VERSION}-ubuntu22.04-${ARCH}.deb