Release 1.1

.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/postgresml/pgcat-ci:latest
+      - image: ghcr.io/levkk/pgcat-ci:1.67
         environment:
           RUST_LOG: info
           LLVM_PROFILE_FILE: /tmp/pgcat-%m-%p.profraw
@@ -63,9 +63,6 @@ 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

@@ -59,7 +59,6 @@ admin_password = "admin_pass"
 # session: one server connection per connected client
 # transaction: one server connection per client transaction
 pool_mode = "transaction"
-prepared_statements_cache_size = 500
 
 # If the client doesn't specify, route traffic to
 # this role by default.
@@ -75,10 +74,6 @@ 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.
@@ -139,10 +134,8 @@ 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"
-prepared_statements_cache_size = 500
 
 [pools.simple_db.users.0]
 username = "simple_user"

.circleci/run_tests.sh

@@ -26,7 +26,6 @@ PGPASSWORD=sharding_user pgbench -h 127.0.0.1 -U sharding_user shard1 -i
 PGPASSWORD=sharding_user pgbench -h 127.0.0.1 -U sharding_user shard2 -i
 
 # Start Toxiproxy
-kill -9 $(pgrep toxiproxy) || true
 LOG_LEVEL=error toxiproxy-server &
 sleep 1
 
@@ -107,26 +106,10 @@ cd ../..
 # These tests will start and stop the pgcat server so it will need to be restarted after the tests
 #
 pip3 install -r tests/python/requirements.txt
-pytest || exit 1
+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
@@ -178,6 +161,3 @@ killall pgcat -s SIGINT
 
 # Allow for graceful shutdown
 sleep 1
-
-kill -9 $(pgrep toxiproxy)
-sleep 1

.github/dependabot.yml

@@ -10,7 +10,3 @@ updates:
     commit-message:
       prefix: "chore(deps)"
     open-pull-requests-limit: 10
-  - package-ecosystem: "github-actions"
-    directory: "/"
-    schedule:
-      interval: "weekly"

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

@@ -2,8 +2,6 @@ name: Build and Push
 
 on:
   push:
-    paths:
-      - '!charts/**.md'
     branches:    
       - main
     tags:
@@ -23,40 +21,33 @@ jobs:
 
     steps:
       - name: Checkout Repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v3
-
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v2
 
       - name: Determine tags
         id: metadata
-        uses: docker/metadata-action@v5
+        uses: docker/metadata-action@v4
         with:
           images: ${{ env.registry }}/${{ env.image-name }}
           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 }}
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v3
+        uses: docker/login-action@v2.1.0
         with:
           registry: ${{ env.registry }}
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Build and push ${{ env.image-name }}
-        uses: docker/build-push-action@v6
+        uses: docker/build-push-action@v3
         with:
-          context: .
-          platforms: linux/amd64,linux/arm64
-          provenance: false
           push: true
           tags: ${{ steps.metadata.outputs.tags }}
           labels: ${{ steps.metadata.outputs.labels }}

.github/workflows/chart-lint-test.yaml

@@ -1,50 +0,0 @@
-name: Lint and Test Charts
-
-on:
-  pull_request:
-    paths:
-      - charts/**
-      - '!charts/**.md'
-jobs:
-  lint-test:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3.1.0
-        with:
-          fetch-depth: 0
-
-      - name: Set up Helm
-        uses: azure/setup-helm@v3
-        with:
-          version: v3.8.1
-
-      # Python is required because `ct lint` runs Yamale (https://github.com/23andMe/Yamale) and
-      # yamllint (https://github.com/adrienverge/yamllint) which require Python
-      - name: Set up Python
-        uses: actions/setup-python@v5.3.0
-        with:
-          python-version: 3.7
-
-      - name: Set up chart-testing
-        uses: helm/chart-testing-action@v2.2.1
-        with:
-          version: v3.5.1
-
-      - name: Run chart-testing (list-changed)
-        id: list-changed
-        run: |
-          changed=$(ct list-changed --config ct.yaml)
-          if [[ -n "$changed" ]]; then
-            echo "changed=true" >> $GITHUB_OUTPUT
-          fi
-
-      - name: Run chart-testing (lint)
-        run: ct lint --config ct.yaml
-
-      - name: Create kind cluster
-        uses: helm/kind-action@v1.10.0
-        if: steps.list-changed.outputs.changed == 'true'
-
-      - name: Run chart-testing (install)
-        run: ct install --config ct.yaml

.github/workflows/chart-release.yaml

@@ -1,40 +0,0 @@
-name: Release Charts
-
-on:
-  push:
-    paths:
-      - charts/**
-      - '!**.md'
-    branches:
-      - main
-
-jobs:
-  release:
-    runs-on: ubuntu-latest
-
-    permissions:
-      contents: write
-
-    steps:
-      - name: Checkout
-        uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608 # v4.1.0
-        with:
-          fetch-depth: 0
-
-      - name: Configure Git
-        run: |
-          git config user.name "$GITHUB_ACTOR"
-          git config user.email "$GITHUB_ACTOR@users.noreply.github.com"
-
-      - name: Install Helm
-        uses: azure/setup-helm@5119fcb9089d432beecbf79bb2c7915207344b78 # v3.5
-        with:
-          version: v3.13.0
-
-      - name: Run chart-releaser
-        uses: helm/chart-releaser-action@a917fd15b20e8b64b94d9158ad54cd6345335584 # v1.6.0
-        with:
-          charts_dir: charts
-          config: cr.yaml
-        env:
-          CR_TOKEN: "${{ secrets.GITHUB_TOKEN }}"

.github/workflows/generate-chart-readme.yaml

@@ -1,48 +0,0 @@
-name: '[CI/CD] Update README metadata'
-
-on:
-  pull_request_target:
-    branches:
-      - main
-    paths:
-      - 'charts/*/values.yaml'
-# Remove all permissions by default
-permissions: {}
-jobs:
-  update-readme-metadata:
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-    steps:
-      - name: Install readme-generator-for-helm
-        run: npm install -g @bitnami/readme-generator-for-helm
-      - name: Checkout
-        uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608
-        with:
-          path: charts
-          ref: ${{github.event.pull_request.head.ref}}
-          repository: ${{github.event.pull_request.head.repo.full_name}}
-          token: ${{ secrets.GITHUB_TOKEN }}
-      - name: Execute readme-generator-for-helm
-        env:
-          DIFF_URL: "${{github.event.pull_request.diff_url}}"
-          TEMP_FILE: "${{runner.temp}}/pr-${{github.event.number}}.diff"
-        run: |
-          # This request doesn't consume API calls.
-          curl -Lkso $TEMP_FILE $DIFF_URL
-          files_changed="$(sed -nr 's/[\-\+]{3} [ab]\/(.*)/\1/p' $TEMP_FILE | sort | uniq)"
-          # Adding || true to avoid "Process exited with code 1" errors
-          charts_dirs_changed="$(echo "$files_changed" | xargs dirname | grep -o "pgcat/[^/]*" | sort | uniq || true)"
-          for chart in ${charts_dirs_changed}; do
-            echo "Updating README.md for ${chart}"
-            readme-generator --values "charts/${chart}/values.yaml" --readme "charts/${chart}/README.md" --schema "/tmp/schema.json"
-          done
-      - name: Push changes
-        run: |
-          # Push all the changes
-          cd charts
-          if git status -s | grep pgcat; then
-            git config user.name "$GITHUB_ACTOR"
-            git config user.email "$GITHUB_ACTOR@users.noreply.github.com"
-            git add . && git commit -am "Update README.md with readme-generator-for-helm" --signoff && git push
-          fi

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

@@ -1,59 +0,0 @@
-name: pgcat package (deb)
-
-on:
-  push:
-    tags:
-      - v*
-  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
-    - name: Set package version
-      if: github.event_name == 'push'  # For push event
-      run: |
-        TAG=${{ github.ref_name }}
-        echo "packageVersion=${TAG#v}" >> "$GITHUB_ENV"
-    - name: Set package version (manual dispatch)
-      if: github.event_name == 'workflow_dispatch'  # For manual dispatch
-      run: echo "packageVersion=${{ github.event.inputs.packageVersion }}" >> "$GITHUB_ENV"
-    - 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 ${{ env.packageVersion }}
-
-        deb-s3 upload \
-          --lock \
-          --bucket apt.postgresml.org \
-          pgcat-${{ env.packageVersion }}-ubuntu22.04-${ARCH}.deb \
-          --codename $(lsb_release -cs)

.gitignore

@@ -10,6 +10,3 @@ lcov.info
 dev/.bash_history
 dev/cache
 !dev/cache/.keepme
-.venv
-**/__pycache__
-.bundle

CONFIG.md

@@ -36,11 +36,10 @@ Port at which prometheus exporter listens on.
 ### connect_timeout
 ```
 path: general.connect_timeout
-default: 1000 # milliseconds
+default: 5000 # milliseconds
 ```
 
-How long the client waits to obtain a server connection before aborting (ms).
+How long to wait before aborting a server connection (ms).
-This is similar to PgBouncer's `query_wait_timeout`.
 
 ### idle_timeout
 ```
@@ -58,38 +57,6 @@ 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
@@ -227,39 +194,6 @@ 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
@@ -296,7 +230,7 @@ default: "random"
 
 Load balancing mode
 `random` selects the server at random
-`loc` selects the server with the least outstanding busy connections
+`loc` selects the server with the least outstanding busy conncetions
 
 ### default_role
 ```
@@ -309,15 +243,6 @@ 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
@@ -463,18 +388,10 @@ path: pools.<pool_name>.users.<user_index>.pool_size
 default: 9
 ```
 
-Maximum number of server connections that can be established for this 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.
 
-### min_pool_size
-```
-path: pools.<pool_name>.users.<user_index>.min_pool_size
-default: 0
-```
-
-Minimum number of idle server connections to retain for this pool.
-
 ### statement_timeout
 ```
 path: pools.<pool_name>.users.<user_index>.statement_timeout
@@ -484,16 +401,6 @@ default: 0
 Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
 0 means it is disabled.
 
-### connect_timeout
-```
-path: pools.<pool_name>.users.<user_index>.connect_timeout
-default: <UNSET> # milliseconds
-```
-
-How long the client waits to obtain a server connection before aborting (ms).
-This is similar to PgBouncer's `query_wait_timeout`.
-If unset, uses the `connect_timeout` defined globally.
-
 ## `pools.<pool_name>.shards.<shard_index>` Section
 
 ### servers
@@ -521,3 +428,4 @@ default: "shard0"
 ```
 
 Database name (e.g. "postgres")
+

CONTRIBUTING.md

@@ -2,36 +2,10 @@
 
 Thank you for contributing! Just a few tips here:
 
-1. `cargo fmt` and `cargo clippy` your code before opening up a PR
+1. `cargo fmt` 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`.
 
-## How to run the integration tests locally and iterate on them
-We have integration tests written in Ruby, Python, Go and Rust.
-Below are the steps to run them in a developer-friendly way that allows iterating and quick turnaround.
-Hear me out, this should be easy, it will involve opening a shell into a container with all the necessary dependancies available for you and you can modify the test code and immediately rerun your test in the interactive shell.
-
-
-Quite simply, make sure you have docker installed and then run
-`./start_test_env.sh`
-
-That is it!
-
-Within this test environment you can modify the file in your favorite IDE and rerun the tests without having to bootstrap the entire environment again.
-
-Once the environment is ready, you can run the tests by running
-Ruby:   `cd /app/tests/ruby && bundle exec ruby <test_name>.rb --format documentation`
-Python: `cd /app/ && pytest`
-Rust:   `cd /app/tests/rust && cargo run`
-Go:     `cd /app/tests/go && /usr/local/go/bin/go test`
-
-You can also rebuild PgCat directly within the environment and the tests will run against the newly built binary
-To rebuild PgCat, just run `cargo build` within the container under `/app`
-
-![Animated gif showing how to run tests](https://github.com/user-attachments/assets/2258fde3-2aed-4efb-bdc5-e4f12dcd4d33)
-
-
-
 Happy hacking!
 
 ## TODOs

Cargo.lock

@@ -17,17 +17,6 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
-[[package]]
-name = "ahash"
-version = "0.8.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f"
-dependencies = [
- "cfg-if",
- "once_cell",
- "version_check",
-]
-
 [[package]]
 name = "aho-corasick"
 version = "1.0.2"
@@ -37,12 +26,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "allocator-api2"
-version = "0.2.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
-
 [[package]]
 name = "android-tzdata"
 version = "0.1.1"
@@ -146,12 +129,6 @@ dependencies = [
  "syn 2.0.26",
 ]
 
-[[package]]
-name = "atomic-waker"
-version = "1.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
-
 [[package]]
 name = "atomic_enum"
 version = "0.2.0"
@@ -192,11 +169,12 @@ checksum = "604178f6c5c21f02dc555784810edfb88d34ac2c73b2eae109655649ee73ce3d"
 
 [[package]]
 name = "bb8"
-version = "0.8.6"
+version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d89aabfae550a5c44b43ab941844ffcd2e993cb6900b342debf59e9ea74acdb8"
+checksum = "98b4b0f25f18bcdc3ac72bdb486ed0acf7e185221fd4dc985bc15db5800b0ba2"
 dependencies = [
  "async-trait",
+ "futures-channel",
  "futures-util",
  "parking_lot",
  "tokio",
@@ -375,6 +353,19 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "env_logger"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85cdab6a89accf66733ad5a1693a4dcced6aeff64602b634530dd73c1f3ee9f0"
+dependencies = [
+ "humantime",
+ "is-terminal",
+ "log",
+ "regex",
+ "termcolor",
+]
+
 [[package]]
 name = "equivalent"
 version = "1.0.1"
@@ -547,32 +538,34 @@ checksum = "b6c80984affa11d98d1b88b66ac8853f143217b399d3c74116778ff8fdb4ed2e"
 
 [[package]]
 name = "h2"
-version = "0.4.6"
+version = "0.3.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "524e8ac6999421f49a846c2d4411f337e53497d8ec55d67753beffa43c5d9205"
+checksum = "97ec8491ebaf99c8eaa73058b045fe58073cd6be7f596ac993ced0b0a0c01049"
 dependencies = [
- "atomic-waker",
  "bytes",
  "fnv",
  "futures-core",
  "futures-sink",
+ "futures-util",
  "http",
- "indexmap",
+ "indexmap 1.9.3",
  "slab",
  "tokio",
  "tokio-util",
  "tracing",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
+
 [[package]]
 name = "hashbrown"
 version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2c6201b9ff9fd90a5a3bac2e56a830d0caa509576f0e503818ee82c181b3437a"
-dependencies = [
- "ahash",
- "allocator-api2",
-]
 
 [[package]]
 name = "heck"
@@ -608,9 +601,9 @@ dependencies = [
 
 [[package]]
 name = "http"
-version = "1.1.0"
+version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21b9ddb458710bc376481b842f5da65cdf31522de232c1ca8146abce2a358258"
+checksum = "bd6effc99afb63425aff9b05836f029929e345a6148a14b7ecd5ab67af944482"
 dependencies = [
  "bytes",
  "fnv",
@@ -619,24 +612,12 @@ dependencies = [
 
 [[package]]
 name = "http-body"
-version = "1.0.1"
+version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
+checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
 dependencies = [
  "bytes",
  "http",
-]
-
-[[package]]
-name = "http-body-util"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "793429d76616a256bcb62c2a2ec2bed781c8307e797e2598c50010f2bee2544f"
-dependencies = [
- "bytes",
- "futures-util",
- "http",
- "http-body",
  "pin-project-lite",
 ]
 
@@ -653,13 +634,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"
 
 [[package]]
-name = "hyper"
+name = "humantime"
-version = "1.4.1"
+version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
+checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
+
+[[package]]
+name = "hyper"
+version = "0.14.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffb1cfd654a8219eaef89881fdb3bb3b1cdc5fa75ded05d6933b2b382e395468"
 dependencies = [
  "bytes",
  "futures-channel",
+ "futures-core",
  "futures-util",
  "h2",
  "http",
@@ -668,26 +656,13 @@ dependencies = [
  "httpdate",
  "itoa",
  "pin-project-lite",
- "smallvec",
+ "socket2 0.4.9",
  "tokio",
+ "tower-service",
+ "tracing",
  "want",
 ]
 
-[[package]]
-name = "hyper-util"
-version = "0.1.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cde7055719c54e36e95e8719f95883f22072a48ede39db7fc17a4e1d5281e9b9"
-dependencies = [
- "bytes",
- "futures-util",
- "http",
- "http-body",
- "hyper",
- "pin-project-lite",
- "tokio",
-]
-
 [[package]]
 name = "iana-time-zone"
 version = "0.1.57"
@@ -732,6 +707,16 @@ dependencies = [
  "unicode-normalization",
 ]
 
+[[package]]
+name = "indexmap"
+version = "1.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99"
+dependencies = [
+ "autocfg",
+ "hashbrown 0.12.3",
+]
+
 [[package]]
 name = "indexmap"
 version = "2.0.0"
@@ -739,7 +724,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d5477fe2230a79769d8dc68e0eabf5437907c0457a5614a9e8dddb67f65eb65d"
 dependencies = [
  "equivalent",
- "hashbrown",
+ "hashbrown 0.14.0",
 ]
 
 [[package]]
@@ -855,15 +840,6 @@ version = "0.4.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b06a4cde4c0f271a446782e3eff8de789548ce57dbc8eca9292c27f4a42004b4"
 
-[[package]]
-name = "lru"
-version = "0.12.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1efa59af2ddfad1854ae27d75009d538d0998b4b2fd47083e743ac1a10e46c60"
-dependencies = [
- "hashbrown",
-]
-
 [[package]]
 name = "lru-cache"
 version = "0.1.2"
@@ -879,15 +855,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ffbee8634e0d45d258acb448e7eaab3fce7a0a467395d4d9f228e3c1f01fb2e4"
 
-[[package]]
-name = "matchers"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558"
-dependencies = [
- "regex-automata 0.1.10",
-]
-
 [[package]]
 name = "matches"
 version = "0.1.10"
@@ -1033,7 +1000,7 @@ checksum = "9b2a4787296e9989611394c33f193f676704af1686e70b8f8033ab5ba9a35a94"
 
 [[package]]
 name = "pgcat"
-version = "1.2.0"
+version = "1.1.0"
 dependencies = [
  "arc-swap",
  "async-trait",
@@ -1043,17 +1010,15 @@ dependencies = [
  "bytes",
  "chrono",
  "clap",
+ "env_logger",
  "exitcode",
  "fallible-iterator",
  "futures",
  "hmac",
- "http-body-util",
  "hyper",
- "hyper-util",
  "itertools",
  "jemallocator",
  "log",
- "lru",
  "md-5",
  "nix",
  "num_cpus",
@@ -1253,17 +1218,8 @@ checksum = "b2eae68fc220f7cf2532e4494aded17545fce192d59cd996e0fe7887f4ceb575"
 dependencies = [
  "aho-corasick",
  "memchr",
- "regex-automata 0.3.3",
+ "regex-automata",
- "regex-syntax 0.7.4",
+ "regex-syntax",
-]
-
-[[package]]
-name = "regex-automata"
-version = "0.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132"
-dependencies = [
- "regex-syntax 0.6.29",
 ]
 
 [[package]]
@@ -1274,15 +1230,9 @@ checksum = "39354c10dd07468c2e73926b23bb9c2caca74c5501e38a35da70406f1d923310"
 dependencies = [
  "aho-corasick",
  "memchr",
- "regex-syntax 0.7.4",
+ "regex-syntax",
 ]
 
-[[package]]
-name = "regex-syntax"
-version = "0.6.29"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1"
-
 [[package]]
 name = "regex-syntax"
 version = "0.7.4"
@@ -1356,9 +1306,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-webpki"
-version = "0.100.2"
+version = "0.100.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e98ff011474fa39949b7e5c0428f9b4937eda7da7848bbb947786b7be0b27dab"
+checksum = "d6207cd5ed3d8dca7816f8f3725513a34609c0c765bf652b8c3cb4cfd87db46b"
 dependencies = [
  "ring",
  "untrusted",
@@ -1493,9 +1443,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.13.2"
+version = "1.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
+checksum = "62bb4feee49fdd9f707ef802e22365a35de4b7b299de4763d44bfea899442ff9"
 
 [[package]]
 name = "socket2"
@@ -1525,9 +1475,9 @@ checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"
 
 [[package]]
 name = "sqlparser"
-version = "0.41.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5cc2c25a6c66789625ef164b4c7d2e548d627902280c13710d33da8222169964"
+checksum = "37d3706eefb17039056234df6b566b0014f303f867f2656108334a55b8096f59"
 dependencies = [
  "log",
  "sqlparser_derive",
@@ -1535,13 +1485,13 @@ dependencies = [
 
 [[package]]
 name = "sqlparser_derive"
-version = "0.2.2"
+version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "01b2e185515564f15375f593fb966b5718bc624ba77fe49fa4616ad619690554"
+checksum = "55fe75cb4a364c7f7ae06c7dbbc8d84bddd85d6cdf9975963c3935bc1991761e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.26",
+ "syn 1.0.109",
 ]
 
 [[package]]
@@ -1594,6 +1544,15 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "termcolor"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be55cf8942feac5c765c2c993422806843c9a9a45d4d5c407ad6dd2ea95eb9b6"
+dependencies = [
+ "winapi-util",
+]
+
 [[package]]
 name = "thiserror"
 version = "1.0.43"
@@ -1756,13 +1715,19 @@ version = "0.19.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8123f27e969974a3dfba720fdb560be359f57b44302d280ba72e76a74480e8a"
 dependencies = [
- "indexmap",
+ "indexmap 2.0.0",
  "serde",
  "serde_spanned",
  "toml_datetime",
  "winnow",
 ]
 
+[[package]]
+name = "tower-service"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6bc1c9ce2b5135ac7f93c72918fc37feb872bdc6a5533a8b85eb4b86bfdae52"
+
 [[package]]
 name = "tracing"
 version = "0.1.37"
@@ -1823,16 +1788,12 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "30a651bc37f915e81f087d86e62a18eec5f79550c7faff886f7090b4ea757c77"
 dependencies = [
- "matchers",
  "nu-ansi-term",
- "once_cell",
- "regex",
  "serde",
  "serde_json",
  "sharded-slab",
  "smallvec",
  "thread_local",
- "tracing",
  "tracing-core",
  "tracing-log",
  "tracing-serde",
@@ -2042,7 +2003,7 @@ version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b03058f88386e5ff5310d9111d53f48b17d732b401aeb83a8d5190f2ac459338"
 dependencies = [
- "rustls-webpki 0.100.2",
+ "rustls-webpki 0.100.1",
 ]
 
 [[package]]
@@ -2067,6 +2028,15 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
 
+[[package]]
+name = "winapi-util"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "winapi-x86_64-pc-windows-gnu"
 version = "0.4.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgcat"
-version = "1.2.0"
+version = "1.1.0"
 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.6"
+bb8 = "0.8.1"
 async-trait = "0.1"
 rand = "0.8"
 chrono = "0.4"
@@ -19,9 +19,10 @@ serde_derive = "1"
 regex = "1"
 num_cpus = "1"
 once_cell = "1"
-sqlparser = { version = "0.41", 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"
@@ -29,9 +30,7 @@ base64 = "0.21"
 stringprep = "0.1"
 tokio-rustls = "0.24"
 rustls-pemfile = "1"
-http-body-util = "0.1.2"
+hyper = { version = "0.14", features = ["full"] }
-hyper = { version = "1.4.1", features = ["full"] }
-hyper-util = { version = "0.1.7", features = ["tokio"] }
 phf = { version = "0.11.1", features = ["macros"] }
 exitcode = "1.1.2"
 futures = "0.3"
@@ -49,12 +48,8 @@ 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 = [
+tracing-subscriber = { version = "0.3.17", features = ["json"]}
-    "json",
-    "env-filter",
-    "std",
-] }
-lru = "0.12.0"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 jemallocator = "0.5.0"
+

Dockerfile

@@ -1,22 +1,11 @@
-FROM rust:1.79.0-slim-bookworm AS builder
+FROM rust:1 AS builder
-
-RUN apt-get update && \
-    apt-get install -y build-essential
-
 COPY . /app
 WORKDIR /app
 RUN cargo build --release
 
-FROM debian:bookworm-slim
+FROM debian:bullseye-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
 ENV RUST_LOG=info
 CMD ["pgcat"]
-STOPSIGNAL SIGINT

Dockerfile.ci

@@ -1,6 +1,4 @@
-FROM cimg/rust:1.79.0
+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 \
@@ -9,9 +7,6 @@ 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

@@ -1,25 +0,0 @@
-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-1-million-requests-per-second">
+      <a href="https://postgresml.org/blog/scaling-postgresml-to-one-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-1-million-requests-per-second">
+      <a href="https://postgresml.org/blog/scaling-postgresml-to-one-million-requests-per-second">
         PostgresML
       </a>
     </td>
@@ -268,8 +268,6 @@ psql -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW DATABASES'
 
 Additionally, Prometheus statistics are available at `/metrics` via HTTP.
 
-We also have a [basic Grafana dashboard](https://github.com/postgresml/pgcat/blob/main/grafana_dashboard.json) based on Prometheus metrics that you can import into Grafana and build on it or use it for monitoring.
-
 ### Live configuration reloading
 
 The config can be reloaded by sending a `kill -s SIGHUP` to the process or by querying `RELOAD` to the admin database. All settings except the `host` and `port` can be reloaded without restarting the pooler, including sharding and replicas configurations.

charts/pgcat/.helmignore

@@ -1,23 +0,0 @@
-# Patterns to ignore when building packages.
-# This supports shell glob matching, relative path matching, and
-# negation (prefixed with !). Only one pattern per line.
-.DS_Store
-# Common VCS dirs
-.git/
-.gitignore
-.bzr/
-.bzrignore
-.hg/
-.hgignore
-.svn/
-# Common backup files
-*.swp
-*.bak
-*.tmp
-*.orig
-*~
-# Various IDEs
-.project
-.idea/
-*.tmproj
-.vscode/

charts/pgcat/Chart.yaml

@@ -1,8 +0,0 @@
-apiVersion: v2
-name: pgcat
-description: A Helm chart for PgCat a PostgreSQL pooler and proxy (like PgBouncer) with support for sharding, load balancing, failover and mirroring.
-maintainers:
-  - name: Wildcard
-    email: support@w6d.io
-appVersion: "1.2.0"
-version: 0.2.1

charts/pgcat/templates/NOTES.txt

@@ -1,22 +0,0 @@
-1. Get the application URL by running these commands:
-{{- if .Values.ingress.enabled }}
-{{- range $host := .Values.ingress.hosts }}
-  {{- range .paths }}
-  http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
-  {{- end }}
-{{- end }}
-{{- else if contains "NodePort" .Values.service.type }}
-  export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "pgcat.fullname" . }})
-  export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
-  echo http://$NODE_IP:$NODE_PORT
-{{- else if contains "LoadBalancer" .Values.service.type }}
-     NOTE: It may take a few minutes for the LoadBalancer IP to be available.
-           You can watch the status of by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "pgcat.fullname" . }}'
-  export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "pgcat.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
-  echo http://$SERVICE_IP:{{ .Values.service.port }}
-{{- else if contains "ClusterIP" .Values.service.type }}
-  export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "pgcat.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
-  export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
-  echo "Visit http://127.0.0.1:8080 to use your application"
-  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
-{{- end }}

charts/pgcat/templates/_config.tpl

@@ -1,3 +0,0 @@
-{{/*
-  Configuration template definition
-*/}}

charts/pgcat/templates/_helpers.tpl

@@ -1,62 +0,0 @@
-{{/*
-Expand the name of the chart.
-*/}}
-{{- define "pgcat.name" -}}
-{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
-{{- end }}
-
-{{/*
-Create a default fully qualified app name.
-We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
-If release name contains chart name it will be used as a full name.
-*/}}
-{{- define "pgcat.fullname" -}}
-{{- if .Values.fullnameOverride }}
-{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
-{{- else }}
-{{- $name := default .Chart.Name .Values.nameOverride }}
-{{- if contains $name .Release.Name }}
-{{- .Release.Name | trunc 63 | trimSuffix "-" }}
-{{- else }}
-{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
-{{- end }}
-{{- end }}
-{{- end }}
-
-{{/*
-Create chart name and version as used by the chart label.
-*/}}
-{{- define "pgcat.chart" -}}
-{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
-{{- end }}
-
-{{/*
-Common labels
-*/}}
-{{- define "pgcat.labels" -}}
-helm.sh/chart: {{ include "pgcat.chart" . }}
-{{ include "pgcat.selectorLabels" . }}
-{{- if .Chart.AppVersion }}
-app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
-{{- end }}
-app.kubernetes.io/managed-by: {{ .Release.Service }}
-{{- end }}
-
-{{/*
-Selector labels
-*/}}
-{{- define "pgcat.selectorLabels" -}}
-app.kubernetes.io/name: {{ include "pgcat.name" . }}
-app.kubernetes.io/instance: {{ .Release.Name }}
-{{- end }}
-
-{{/*
-Create the name of the service account to use
-*/}}
-{{- define "pgcat.serviceAccountName" -}}
-{{- if .Values.serviceAccount.create }}
-{{- default (include "pgcat.fullname" .) .Values.serviceAccount.name }}
-{{- else }}
-{{- default "default" .Values.serviceAccount.name }}
-{{- end }}
-{{- end }}

charts/pgcat/templates/deployment.yaml

@@ -1,66 +0,0 @@
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: {{ include "pgcat.fullname" . }}
-  labels:
-    {{- include "pgcat.labels" . | nindent 4 }}
-spec:
-  replicas: {{ .Values.replicaCount }}
-  selector:
-    matchLabels:
-      {{- include "pgcat.selectorLabels" . | nindent 6 }}
-  template:
-    metadata:
-      annotations:
-        checksum/secret: {{ include (print $.Template.BasePath "/secret.yaml") . | sha256sum }}
-      {{- with .Values.podAnnotations }}
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      labels:
-        {{- include "pgcat.selectorLabels" . | nindent 8 }}
-    spec:
-      {{- with .Values.image.pullSecrets }}
-      imagePullSecrets:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      serviceAccountName: {{ include "pgcat.serviceAccountName" . }}
-      securityContext:
-        {{- toYaml .Values.podSecurityContext | nindent 8 }}
-      containers:
-        - name: {{ .Chart.Name }}
-          securityContext:
-            {{- toYaml .Values.containerSecurityContext | nindent 12 }}
-          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-          imagePullPolicy: {{ .Values.image.pullPolicy }}
-          ports:
-            - name: pgcat
-              containerPort: {{ .Values.configuration.general.port }}
-              protocol: TCP
-          livenessProbe:
-            tcpSocket:
-              port: pgcat
-          readinessProbe:
-            tcpSocket:
-              port: pgcat
-          resources:
-            {{- toYaml .Values.resources | nindent 12 }}
-          volumeMounts:
-          - mountPath: /etc/pgcat
-            name: config
-      {{- with .Values.nodeSelector }}
-      nodeSelector:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      {{- with .Values.tolerations }}
-      tolerations:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      volumes:
-      - secret:
-          defaultMode: 420
-          secretName: {{ include "pgcat.fullname" . }}
-        name: config

charts/pgcat/templates/ingress.yaml

@@ -1,61 +0,0 @@
-{{- if .Values.ingress.enabled -}}
-{{- $fullName := include "pgcat.fullname" . -}}
-{{- $svcPort := .Values.service.port -}}
-{{- if and .Values.ingress.className (not (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion)) }}
-  {{- if not (hasKey .Values.ingress.annotations "kubernetes.io/ingress.class") }}
-  {{- $_ := set .Values.ingress.annotations "kubernetes.io/ingress.class" .Values.ingress.className}}
-  {{- end }}
-{{- end }}
-{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion -}}
-apiVersion: networking.k8s.io/v1
-{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion -}}
-apiVersion: networking.k8s.io/v1beta1
-{{- else -}}
-apiVersion: extensions/v1beta1
-{{- end }}
-kind: Ingress
-metadata:
-  name: {{ $fullName }}
-  labels:
-    {{- include "pgcat.labels" . | nindent 4 }}
-  {{- with .Values.ingress.annotations }}
-  annotations:
-    {{- toYaml . | nindent 4 }}
-  {{- end }}
-spec:
-  {{- if and .Values.ingress.className (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion) }}
-  ingressClassName: {{ .Values.ingress.className }}
-  {{- end }}
-  {{- if .Values.ingress.tls }}
-  tls:
-    {{- range .Values.ingress.tls }}
-    - hosts:
-        {{- range .hosts }}
-        - {{ . | quote }}
-        {{- end }}
-      secretName: {{ .secretName }}
-    {{- end }}
-  {{- end }}
-  rules:
-    {{- range .Values.ingress.hosts }}
-    - host: {{ .host | quote }}
-      http:
-        paths:
-          {{- range .paths }}
-          - path: {{ .path }}
-            {{- if and .pathType (semverCompare ">=1.18-0" $.Capabilities.KubeVersion.GitVersion) }}
-            pathType: {{ .pathType }}
-            {{- end }}
-            backend:
-              {{- if semverCompare ">=1.19-0" $.Capabilities.KubeVersion.GitVersion }}
-              service:
-                name: {{ $fullName }}
-                port:
-                  number: {{ $svcPort }}
-              {{- else }}
-              serviceName: {{ $fullName }}
-              servicePort: {{ $svcPort }}
-              {{- end }}
-          {{- end }}
-    {{- end }}
-{{- end }}

charts/pgcat/templates/secret.yaml

@@ -1,97 +0,0 @@
-apiVersion: v1
-kind: Secret
-metadata:
-  name: {{ include "pgcat.fullname" . }}
-  labels:
-    {{- include "pgcat.labels" . | nindent 4 }}
-type: Opaque
-stringData:
-  pgcat.toml: |
-    [general]
-    host = {{ .Values.configuration.general.host | quote }}
-    port = {{ .Values.configuration.general.port }}
-    enable_prometheus_exporter = {{ .Values.configuration.general.enable_prometheus_exporter }}
-    prometheus_exporter_port = {{ .Values.configuration.general.prometheus_exporter_port }}
-    connect_timeout = {{ .Values.configuration.general.connect_timeout }}
-    idle_timeout = {{ .Values.configuration.general.idle_timeout | int }}
-    server_lifetime = {{ .Values.configuration.general.server_lifetime | int }}
-    server_tls = {{ .Values.configuration.general.server_tls }}
-    idle_client_in_transaction_timeout = {{ .Values.configuration.general.idle_client_in_transaction_timeout | int }}
-    healthcheck_timeout = {{ .Values.configuration.general.healthcheck_timeout }}
-    healthcheck_delay = {{ .Values.configuration.general.healthcheck_delay }}
-    shutdown_timeout = {{ .Values.configuration.general.shutdown_timeout }}
-    ban_time = {{ .Values.configuration.general.ban_time }}
-    log_client_connections = {{ .Values.configuration.general.log_client_connections }}
-    log_client_disconnections = {{ .Values.configuration.general.log_client_disconnections }}
-    tcp_keepalives_idle = {{ .Values.configuration.general.tcp_keepalives_idle }}
-    tcp_keepalives_count = {{ .Values.configuration.general.tcp_keepalives_count }}
-    tcp_keepalives_interval = {{ .Values.configuration.general.tcp_keepalives_interval }}
-    {{- if and (ne .Values.configuration.general.tls_certificate "-") (ne .Values.configuration.general.tls_private_key "-") }}
-    tls_certificate = "{{ .Values.configuration.general.tls_certificate }}"
-    tls_private_key = "{{ .Values.configuration.general.tls_private_key }}"
-    {{- end }}
-    admin_username = {{ .Values.configuration.general.admin_username | quote }}
-    admin_password = {{ .Values.configuration.general.admin_password | quote }}
-    {{- if and .Values.configuration.general.auth_query_user .Values.configuration.general.auth_query_password .Values.configuration.general.auth_query }}
-    auth_query = {{ .Values.configuration.general.auth_query | quote }}
-    auth_query_user = {{ .Values.configuration.general.auth_query_user | quote }}
-    auth_query_password = {{ .Values.configuration.general.auth_query_password | quote }}
-    {{- end }}
-
-    {{- range $pool := .Values.configuration.pools }}
-
-    ##
-    ## pool for {{ $pool.name }}
-    ##
-    [pools.{{ $pool.name | quote }}]
-    pool_mode = {{ default "transaction" $pool.pool_mode | quote }}
-    load_balancing_mode = {{ default "random" $pool.load_balancing_mode | quote }}
-    default_role = {{ default "any" $pool.default_role | quote }}
-    prepared_statements_cache_size = {{ default 500 $pool.prepared_statements_cache_size }}
-    query_parser_enabled = {{ default true $pool.query_parser_enabled }}
-    query_parser_read_write_splitting = {{ default true $pool.query_parser_read_write_splitting }}
-    primary_reads_enabled = {{ default true $pool.primary_reads_enabled }}
-    sharding_function = {{ default "pg_bigint_hash" $pool.sharding_function | quote }}
-
-    {{-   range $index, $user := $pool.users }}
-
-    ## pool {{ $pool.name }} user {{ $user.username | quote }}
-    ##
-    [pools.{{ $pool.name | quote }}.users.{{ $index }}]
-    username = {{ $user.username | quote }}
-    {{- if $user.password }}
-    password = {{ $user.password | quote }}
-    {{- else if and $user.passwordSecret.name $user.passwordSecret.key }}
-    {{- $secret := (lookup "v1" "Secret" $.Release.Namespace $user.passwordSecret.name) }}
-    {{- if $secret }}
-    {{- $password := index $secret.data $user.passwordSecret.key | b64dec }}
-    password = {{ $password | quote }}
-    {{- end }}
-    {{- end }}
-    pool_size = {{ $user.pool_size }}
-    statement_timeout = {{ default 0 $user.statement_timeout }}
-    min_pool_size = {{ default 3 $user.min_pool_size }}
-    {{- if $user.server_lifetime }}
-    server_lifetime = {{ $user.server_lifetime }}
-    {{- end }}
-    {{-     if and $user.server_username $user.server_password }}
-    server_username = {{ $user.server_username | quote }}
-    server_password = {{ $user.server_password | quote }}
-    {{-     end }}
-    {{-   end }}
-
-    {{-   range $index, $shard := $pool.shards }}
-
-    ## pool {{ $pool.name }} database {{ $shard.database }}
-    ##
-    [pools.{{ $pool.name | quote }}.shards.{{ $index }}]
-    {{-     if gt (len $shard.servers) 0}}
-    servers = [
-    {{-       range  $server := $shard.servers }}
-        [ {{ $server.host | quote }}, {{ $server.port }}, {{ $server.role | quote }} ],
-    {{-       end }}
-    ]
-    {{-     end }}
-    database = {{ $shard.database | quote }}
-    {{-   end }}
-    {{- end }}

charts/pgcat/templates/service.yaml

@@ -1,15 +0,0 @@
-apiVersion: v1
-kind: Service
-metadata:
-  name: {{ include "pgcat.fullname" . }}
-  labels:
-    {{- include "pgcat.labels" . | nindent 4 }}
-spec:
-  type: {{ .Values.service.type }}
-  ports:
-    - port: {{ .Values.service.port }}
-      targetPort: pgcat
-      protocol: TCP
-      name: pgcat
-  selector:
-    {{- include "pgcat.selectorLabels" . | nindent 4 }}

charts/pgcat/templates/serviceaccount.yaml

@@ -1,12 +0,0 @@
-{{- if .Values.serviceAccount.create -}}
-apiVersion: v1
-kind: ServiceAccount
-metadata:
-  name: {{ include "pgcat.serviceAccountName" . }}
-  labels:
-    {{- include "pgcat.labels" . | nindent 4 }}
-  {{- with .Values.serviceAccount.annotations }}
-  annotations:
-    {{- toYaml . | nindent 4 }}
-  {{- end }}
-{{- end }}

charts/pgcat/values.yaml

@@ -1,374 +0,0 @@
-## String to partially override aspnet-core.fullname template (will maintain the release name)
-## @param nameOverride String to partially override common.names.fullname
-##
-nameOverride: ""
-
-## String to fully override aspnet-core.fullname template
-## @param fullnameOverride String to fully override common.names.fullname
-##
-fullnameOverride: ""
-
-## Number of PgCat replicas to deploy
-## @param replicaCount Number of PgCat replicas to deploy
-replicaCount: 1
-
-## Bitnami PgCat image version
-## ref: https://hub.docker.com/r/bitnami/kubewatch/tags/
-##
-## @param image.registry PgCat image registry
-## @param image.repository PgCat image name
-## @param image.tag PgCat image tag
-## @param image.pullPolicy PgCat image tag
-## @param image.pullSecrets Specify docker-registry secret names as an array
-image:
-  repository: ghcr.io/postgresml/pgcat
-  # Overrides the image tag whose default is the chart appVersion.
-  tag: "main"
-  ## Specify a imagePullPolicy
-  ## Defaults to 'Always' if image tag is 'latest', else set to 'IfNotPresent'
-  ## ref: http://kubernetes.io/docs/user-guide/images/#pre-pulling-images
-  ##
-  pullPolicy: IfNotPresent
-  ## Optionally specify an array of imagePullSecrets.
-  ## Secrets must be manually created in the namespace.
-  ## ref: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
-  ## Example:
-  ## pullSecrets:
-  ##   - myRegistryKeySecretName
-  ##
-  pullSecrets: []
-
-## Specifies whether a ServiceAccount should be created
-##
-## @param serviceAccount.create Enable the creation of a ServiceAccount for PgCat pods
-## @param serviceAccount.name Name of the created ServiceAccount
-##
-serviceAccount:
-  ## Specifies whether a service account should be created
-  create: true
-  ## Annotations to add to the service account
-  annotations: {}
-  ## The name of the service account to use.
-  ## If not set and create is true, a name is generated using the fullname template
-  name: ""
-
-## Annotations for server pods.
-## ref: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
-##
-## @param podAnnotations  Annotations for PgCat pods
-##
-podAnnotations: {}
-
-## PgCat containers' SecurityContext
-## ref: https://kubernetes.io/docs/tasks/configure-pod-container/security-context/#set-the-security-context-for-a-pod
-##
-## @param podSecurityContext.enabled Enabled PgCat pods' Security Context
-## @param podSecurityContext.fsGroup Set PgCat pod's Security Context fsGroup
-##
-podSecurityContext: {}
-  # fsGroup: 2000
-
-## PgCat pods' Security Context
-## ref: https://kubernetes.io/docs/tasks/configure-pod-container/security-context/#set-the-security-context-for-a-container
-##
-## @param containerSecurityContext.enabled Enabled PgCat containers' Security Context
-## @param containerSecurityContext.runAsUser Set PgCat container's Security Context runAsUser
-## @param containerSecurityContext.runAsNonRoot Set PgCat container's Security Context runAsNonRoot
-##
-containerSecurityContext: {}
-  # capabilities:
-  #   drop:
-  #   - ALL
-  # readOnlyRootFilesystem: true
-  # runAsNonRoot: true
-  # runAsUser: 1000
-
-## PgCat service
-##
-## @param service.type PgCat service type
-## @param service.port PgCat service port
-service:
-  type: ClusterIP
-  port: 6432
-
-ingress:
-  enabled: false
-  className: ""
-  annotations: {}
-    # kubernetes.io/ingress.class: nginx
-    # kubernetes.io/tls-acme: "true"
-  hosts:
-    - host: chart-example.local
-      paths:
-        - path: /
-          pathType: ImplementationSpecific
-  tls: []
-  #  - secretName: chart-example-tls
-  #    hosts:
-  #      - chart-example.local
-
-## PgCat resource requests and limits
-## ref: http://kubernetes.io/docs/user-guide/compute-resources/
-##
-## @skip resources Optional description
-## @disabled-param resources.limits  The resources limits for the PgCat container
-## @disabled-param resources.requests  The requested resources for the PgCat container
-##
-resources:
-  # We usually recommend not to specify default resources and to leave this as a conscious
-  # choice for the user. This also increases chances charts run on environments with little
-  # resources, such as Minikube. If you do want to specify resources, uncomment the following
-  # lines, adjust them as necessary, and remove the curly braces after 'resources:'.
-  limits: {}
-  #   cpu: 100m
-  #   memory: 128Mi
-  requests: {}
-  #   cpu: 100m
-  #   memory: 128Mi
-
-## Node labels for pod assignment. Evaluated as a template.
-## ref: https://kubernetes.io/docs/user-guide/node-selection/
-##
-## @param nodeSelector  Node labels for pod assignment
-##
-nodeSelector: {}
-
-## Tolerations for pod assignment. Evaluated as a template.
-## ref: https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/
-##
-## @param tolerations  Tolerations for pod assignment
-##
-tolerations: []
-
-## Affinity for pod assignment. Evaluated as a template.
-## ref: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#affinity-and-anti-affinity
-## Note: podAffinityPreset, podAntiAffinityPreset, and nodeAffinityPreset will be ignored when it's set
-##
-## @param affinity  Affinity for pod assignment
-##
-affinity: {}
-
-## PgCat configuration
-## @param configuration [object]
-configuration:
-  ## General pooler settings
-  ## @param [object]
-  general:
-    ## @param configuration.general.host What IP to run on, 0.0.0.0 means accessible from everywhere.
-    host: "0.0.0.0"
-
-    ## @param configuration.general.port Port to run on, same as PgBouncer used in this example.
-    port: 6432
-
-    ## @param configuration.general.enable_prometheus_exporter Whether to enable prometheus exporter or not.
-    enable_prometheus_exporter: false
-
-    ## @param configuration.general.prometheus_exporter_port Port at which prometheus exporter listens on.
-    prometheus_exporter_port: 9930
-
-    # @param configuration.general.connect_timeout How long to wait before aborting a server connection (ms).
-    connect_timeout: 5000
-
-    # How long an idle connection with a server is left open (ms).
-    idle_timeout: 30000  # milliseconds
-
-    # Max connection lifetime before it's closed, even if actively used.
-    server_lifetime: 86400000  # 24 hours
-
-    # Whether to use TLS for server connections or not.
-    server_tls: false
-
-    # How long a client is allowed to be idle while in a transaction (ms).
-    idle_client_in_transaction_timeout: 0  # milliseconds
-
-    # @param configuration.general.healthcheck_timeout How much time to give `SELECT 1` health check query to return with a result (ms).
-    healthcheck_timeout: 1000
-
-    # @param configuration.general.healthcheck_delay How long to keep connection available for immediate re-use, without running a healthcheck query on it
-    healthcheck_delay: 30000
-
-    # @param configuration.general.shutdown_timeout How much time to give clients during shutdown before forcibly killing client connections (ms).
-    shutdown_timeout: 60000
-
-    # @param configuration.general.ban_time For how long to ban a server if it fails a health check (seconds).
-    ban_time: 60    # seconds
-
-    # @param configuration.general.log_client_connections If we should log client connections
-    log_client_connections: false
-
-    # @param configuration.general.log_client_disconnections If we should log client disconnections
-    log_client_disconnections: false
-
-    # TLS
-    # tls_certificate: "server.cert"
-    # tls_private_key: "server.key"
-    tls_certificate: "-"
-    tls_private_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: "postgres"
-    admin_password: "postgres"
-
-    # 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: null
-
-    # 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.
-    #
-    # @param configuration.general.auth_query_user
-    auth_query_user: null
-
-    # 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.
-    #
-    # @param configuration.general.auth_query_password
-    auth_query_password: null
-
-    # Number of seconds of connection idleness to wait before sending a keepalive packet to the server.
-    tcp_keepalives_idle: 5
-
-    # Number of unacknowledged keepalive packets allowed before giving up and closing the connection.
-    tcp_keepalives_count: 5
-
-    # Number of seconds between keepalive packets.
-    tcp_keepalives_interval: 5
-
-  ## 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"
-  ## @param [object]
-  pools:
-    [{
-      name: "simple", pool_mode: "transaction",
-      users: [{username: "user", password: "pass", pool_size: 5, statement_timeout: 0}],
-      shards: [{
-        servers: [{host: "postgres", port: 5432, role: "primary"}],
-        database: "postgres"
-      }]
-    }]
-    # - ## default values
-    #   ##
-    #   ##
-    #   ##
-    #   name: "db"
-
-    #  ## Pool mode (see PgBouncer docs for more).
-    #  ## session: one server connection per connected client
-    #  ## transaction: one server connection per client transaction
-    #  ## @param configuration.poolsPostgres.pool_mode
-    #  pool_mode: "transaction"
-
-    #  ## Load balancing mode
-    #  ## `random` selects the server at random
-    #  ## `loc` selects the server with the least outstanding busy connections
-    #  ##
-    #  ## @param configuration.poolsPostgres.load_balancing_mode
-    #  load_balancing_mode: "random"
-
-    #  ## Prepared statements cache size.
-    #  ## TODO: update documentation
-    #  ##
-    #  ## @param configuration.poolsPostgres.prepared_statements_cache_size
-    #  prepared_statements_cache_size: 500
-
-    #  ## 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.
-    #  ## @param configuration.poolsPostgres.default_role
-    #  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.
-    #  ## @param configuration.poolsPostgres.query_parser_enabled
-    #  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.
-    #  ## @param configuration.poolsPostgres.query_parser_read_write_splitting
-    #  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.
-    #  ## @param configuration.poolsPostgres.primary_reads_enabled
-    #  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
-    #  ##
-    #  ## @param configuration.poolsPostgres.sharding_function
-    #  sharding_function: "pg_bigint_hash"
-
-    #  ## Credentials for users that may connect to this cluster
-    #  ## @param users [array]
-    #  ## @param users[0].username Name of the env var (required)
-    #  ## @param users[0].password Value for the env var (required) leave empty to use existing secret see passwordSecret.name and passwordSecret.key
-    #  ## @param users[0].passwordSecret.name Name of the secret containing the password
-    #  ## @param users[0].passwordSecret.key Key in the secret containing the password
-    #  ## @param users[0].pool_size Maximum number of server connections that can be established for this user
-    #  ## @param users[0].statement_timeout Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
-    #  users: []
-    #    # - username: "user"
-    #    #   password: "pass"
-    #    #
-    #    #  # 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: 9
-    #    #
-    #    #  # Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
-    #    #  statement_timeout: 0
-    #    #
-    #    #  # PostgreSQL username used to connect to the server.
-    #    #  server_username: "postgres
-    #    #
-    #    #  # PostgreSQL password used to connect to the server.
-    #    #  server_password: "postgres
-
-    #  ## @param shards [array]
-    #  ## @param shards[0].server[0].host Host for this shard
-    #  ## @param shards[0].server[0].port Port for this shard
-    #  ## @param shards[0].server[0].role Role for this shard
-    #  shards: []
-    #      # [ host, port, role ]
-    #    # - servers:
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "primary"
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "replica"
-    #    #   database: "postgres"
-    #    #   # [ host, port, role ]
-    #    # - servers:
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "primary"
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "replica"
-    #    #   database: "postgres"
-    #    #   # [ host, port, role ]
-    #    # - servers:
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "primary"
-    #    #     - host: "postgres"
-    #    #       port: 5432
-    #    #       role: "replica"
-    #    #   database: "postgres"

control

@@ -1,9 +0,0 @@
-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.

cr.yaml

@@ -1 +0,0 @@
-sign: false

ct.yaml

@@ -1,5 +0,0 @@
-remote: origin
-target-branch: main
-chart-dirs:
-  - charts
-

dev/Dockerfile

@@ -1,8 +1,6 @@
-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,10 +71,6 @@ 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

@@ -1,17 +0,0 @@
-[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
-ExecReload=/bin/kill -SIGHUP $MAINPID
-
-[Install]
-WantedBy=multi-user.target

pgcat.toml

@@ -60,6 +60,12 @@ tcp_keepalives_count = 5
 # Number of seconds between keepalive packets.
 tcp_keepalives_interval = 5
 
+# Handle prepared statements.
+prepared_statements = true
+
+# Prepared statements server cache size.
+prepared_statements_cache_size = 500
+
 # Path to TLS Certificate file to use for TLS connections
 # tls_certificate = ".circleci/server.cert"
 # Path to TLS private key file to use for TLS connections
@@ -150,20 +156,12 @@ 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.
@@ -175,12 +173,6 @@ 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:
@@ -191,7 +183,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 usename, passwd FROM pg_shadow WHERE usename='$1'"
+# auth_query = "SELECT $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.
@@ -278,7 +270,7 @@ username = "sharding_user"
 # if `server_password` is not set.
 password = "sharding_user"
 
-pool_mode = "transaction"
+pool_mode = "session"
 
 # PostgreSQL username used to connect to the server.
 # server_username = "another_user"
@@ -301,8 +293,6 @@ 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

@@ -1,13 +0,0 @@
-#!/bin/bash
-set -e
-
-systemctl daemon-reload
-systemctl enable pgcat
-
-if ! id pgcat 2> /dev/null; then
-	useradd -s /usr/bin/false pgcat
-fi
-
-if [ -f /etc/pgcat.toml ]; then
-	systemctl start pgcat
-fi

postrm

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

prerm

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

src/admin.rs

@@ -1,5 +1,4 @@
 use crate::pool::BanReason;
-use crate::server::ServerParameters;
 use crate::stats::pool::PoolStats;
 use bytes::{Buf, BufMut, BytesMut};
 use log::{error, info, trace};
@@ -18,16 +17,16 @@ use crate::pool::ClientServerMap;
 use crate::pool::{get_all_pools, get_pool};
 use crate::stats::{get_client_stats, get_server_stats, ClientState, ServerState};
 
-pub fn generate_server_parameters_for_admin() -> ServerParameters {
+pub fn generate_server_info_for_admin() -> BytesMut {
-    let mut server_parameters = ServerParameters::new();
+    let mut server_info = BytesMut::new();
 
-    server_parameters.set_param("application_name".to_string(), "".to_string(), true);
+    server_info.put(server_parameter_message("application_name", ""));
-    server_parameters.set_param("client_encoding".to_string(), "UTF8".to_string(), true);
+    server_info.put(server_parameter_message("client_encoding", "UTF8"));
-    server_parameters.set_param("server_encoding".to_string(), "UTF8".to_string(), true);
+    server_info.put(server_parameter_message("server_encoding", "UTF8"));
-    server_parameters.set_param("server_version".to_string(), VERSION.to_string(), true);
+    server_info.put(server_parameter_message("server_version", VERSION));
-    server_parameters.set_param("DateStyle".to_string(), "ISO, MDY".to_string(), true);
+    server_info.put(server_parameter_message("DateStyle", "ISO, MDY"));
 
-    server_parameters
+    server_info
 }
 
 /// Handle admin client.
@@ -55,12 +54,7 @@ where
 
     let query_parts: Vec<&str> = query.trim_end_matches(';').split_whitespace().collect();
 
-    match query_parts
+    match query_parts[0].to_ascii_uppercase().as_str() {
-        .first()
-        .unwrap_or(&"")
-        .to_ascii_uppercase()
-        .as_str()
-    {
         "BAN" => {
             trace!("BAN");
             ban(stream, query_parts).await
@@ -79,22 +73,17 @@ where
         }
         "PAUSE" => {
             trace!("PAUSE");
-            pause(stream, query_parts).await
+            pause(stream, query_parts[1]).await
         }
         "RESUME" => {
             trace!("RESUME");
-            resume(stream, query_parts).await
+            resume(stream, query_parts[1]).await
         }
         "SHUTDOWN" => {
             trace!("SHUTDOWN");
             shutdown(stream).await
         }
-        "SHOW" => match query_parts
+        "SHOW" => match query_parts[1].to_ascii_uppercase().as_str() {
-            .get(1)
-            .unwrap_or(&"")
-            .to_ascii_uppercase()
-            .as_str()
-        {
             "HELP" => {
                 trace!("SHOW HELP");
                 show_help(stream).await
@@ -293,7 +282,7 @@ where
 {
     let mut res = BytesMut::new();
 
-    let detail_msg = [
+    let detail_msg = vec![
         "",
         "SHOW HELP|CONFIG|DATABASES|POOLS|CLIENTS|SERVERS|USERS|VERSION",
         // "SHOW PEERS|PEER_POOLS", // missing PEERS|PEER_POOLS
@@ -311,6 +300,7 @@ where
         // "KILL <db>",
         // "SUSPEND",
         "SHUTDOWN",
+        // "WAIT_CLOSE [<db>]", // missing
     ];
 
     res.put(notify("Console usage", detail_msg.join("\n\t")));
@@ -700,8 +690,6 @@ 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();
@@ -709,7 +697,6 @@ 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(),
@@ -723,8 +710,6 @@ 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));
@@ -759,7 +744,6 @@ where
         ("age_seconds", DataType::Numeric),
         ("prepare_cache_hit", DataType::Numeric),
         ("prepare_cache_miss", DataType::Numeric),
-        ("prepare_cache_eviction", DataType::Numeric),
         ("prepare_cache_size", DataType::Numeric),
     ];
 
@@ -792,10 +776,6 @@ where
                 .prepared_miss_count
                 .load(Ordering::Relaxed)
                 .to_string(),
-            server
-                .prepared_eviction_count
-                .load(Ordering::Relaxed)
-                .to_string(),
             server
                 .prepared_cache_size
                 .load(Ordering::Relaxed)
@@ -816,128 +796,96 @@ where
 }
 
 /// Pause a pool. It won't pass any more queries to the backends.
-async fn pause<T>(stream: &mut T, tokens: Vec<&str>) -> Result<(), Error>
+async fn pause<T>(stream: &mut T, query: &str) -> Result<(), Error>
 where
     T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let parts: Vec<&str> = match tokens.len() == 2 {
+    let parts: Vec<&str> = query.split(",").map(|part| part.trim()).collect();
-        true => tokens[1].split(',').map(|part| part.trim()).collect(),
-        false => Vec::new(),
-    };
 
-    match parts.len() {
+    if parts.len() != 2 {
-        0 => {
+        error_response(
-            for (_, pool) in get_all_pools() {
+            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) => {
                 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
             }
 
-            let mut res = BytesMut::new();
+            None => {
-
+                error_response(
-            res.put(command_complete("PAUSE"));
+                    stream,
-
+                    &format!(
-            // ReadyForQuery
+                        "No pool configured for database: {}, user: {}",
-            res.put_u8(b'Z');
+                        database, user
-            res.put_i32(5);
+                    ),
-            res.put_u8(b'I');
+                )
-
+                .await
-            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, tokens: Vec<&str>) -> Result<(), Error>
+async fn resume<T>(stream: &mut T, query: &str) -> Result<(), Error>
 where
     T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let parts: Vec<&str> = match tokens.len() == 2 {
+    let parts: Vec<&str> = query.split(",").map(|part| part.trim()).collect();
-        true => tokens[1].split(',').map(|part| part.trim()).collect(),
-        false => Vec::new(),
-    };
 
-    match parts.len() {
+    if parts.len() != 2 {
-        0 => {
+        error_response(
-            for (_, pool) in get_all_pools() {
+            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) => {
                 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
             }
 
-            let mut res = BytesMut::new();
+            None => {
-
+                error_response(
-            res.put(command_complete("RESUME"));
+                    stream,
-
+                    &format!(
-            // ReadyForQuery
+                        "No pool configured for database: {}, user: {}",
-            res.put_u8(b'Z');
+                        database, user
-            res.put_i32(5);
+                    ),
-            res.put_u8(b'I');
+                )
-
+                .await
-            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

@@ -1,4 +1,3 @@
-use crate::config::AuthType;
 use crate::errors::Error;
 use crate::pool::ConnectionPool;
 use crate::server::Server;
@@ -72,7 +71,6 @@ impl AuthPassthrough {
     pub async fn fetch_hash(&self, address: &crate::config::Address) -> Result<String, Error> {
         let auth_user = crate::config::User {
             username: self.user.clone(),
-            auth_type: AuthType::MD5,
             password: Some(self.password.clone()),
             server_username: None,
             server_password: None,
@@ -81,8 +79,6 @@ 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

@@ -25,7 +25,7 @@ pub struct Args {
 }
 
 pub fn parse() -> Args {
-    Args::parse()
+    return Args::parse();
 }
 
 #[derive(ValueEnum, Clone, Debug)]

src/config.rs

@@ -3,14 +3,11 @@ 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;
@@ -38,12 +35,12 @@ pub enum Role {
     Mirror,
 }
 
-impl std::fmt::Display for Role {
+impl ToString for Role {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    fn to_string(&self) -> String {
-        match self {
+        match *self {
-            Role::Primary => write!(f, "primary"),
+            Role::Primary => "primary".to_string(),
-            Role::Replica => write!(f, "replica"),
+            Role::Replica => "replica".to_string(),
-            Role::Mirror => write!(f, "mirror"),
+            Role::Mirror => "mirror".to_string(),
         }
     }
 }
@@ -104,9 +101,6 @@ 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 {
@@ -116,15 +110,14 @@ 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)),
         }
     }
 }
@@ -189,18 +182,6 @@ 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.
@@ -208,9 +189,6 @@ impl Address {
 pub struct User {
     pub username: String,
     pub password: Option<String>,
-
-    #[serde(default = "User::default_auth_type")]
-    pub auth_type: AuthType,
     pub server_username: Option<String>,
     pub server_password: Option<String>,
     pub pool_size: u32,
@@ -219,8 +197,6 @@ 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 {
@@ -228,7 +204,6 @@ impl Default for User {
         User {
             username: String::from("postgres"),
             password: None,
-            auth_type: AuthType::MD5,
             server_username: None,
             server_password: None,
             pool_size: 15,
@@ -236,26 +211,24 @@ impl Default for User {
             statement_timeout: 0,
             pool_mode: None,
             server_lifetime: None,
-            connect_timeout: None,
-            idle_timeout: None,
         }
     }
 }
 
 impl User {
-    pub fn default_auth_type() -> AuthType {
-        AuthType::MD5
-    }
-
     fn validate(&self) -> Result<(), Error> {
-        if let Some(min_pool_size) = self.min_pool_size {
+        match self.min_pool_size {
-            if min_pool_size > self.pool_size {
+            Some(min_pool_size) => {
-                error!(
+                if min_pool_size > self.pool_size {
-                    "min_pool_size of {} cannot be larger than pool_size of {}",
+                    error!(
-                    min_pool_size, self.pool_size
+                        "min_pool_size of {} cannot be larger than pool_size of {}",
-                );
+                        min_pool_size, self.pool_size
-                return Err(Error::BadConfig);
+                    );
+                    return Err(Error::BadConfig);
+                }
             }
+
+            None => (),
         };
 
         Ok(())
@@ -342,9 +315,6 @@ pub struct General {
     pub admin_username: String,
     pub admin_password: String,
 
-    #[serde(default = "General::default_admin_auth_type")]
-    pub admin_auth_type: AuthType,
-
     #[serde(default = "General::default_validate_config")]
     pub validate_config: bool,
 
@@ -352,6 +322,12 @@ pub struct General {
     pub auth_query: Option<String>,
     pub auth_query_user: Option<String>,
     pub auth_query_password: Option<String>,
+
+    #[serde(default)]
+    pub prepared_statements: bool,
+
+    #[serde(default = "General::default_prepared_statements_cache_size")]
+    pub prepared_statements_cache_size: usize,
 }
 
 impl General {
@@ -359,10 +335,6 @@ impl General {
         "0.0.0.0".into()
     }
 
-    pub fn default_admin_auth_type() -> AuthType {
-        AuthType::MD5
-    }
-
     pub fn default_port() -> u16 {
         5432
     }
@@ -437,6 +409,10 @@ impl General {
     pub fn default_server_round_robin() -> bool {
         true
     }
+
+    pub fn default_prepared_statements_cache_size() -> usize {
+        500
+    }
 }
 
 impl Default for General {
@@ -448,34 +424,35 @@ impl Default for General {
             prometheus_exporter_port: 9930,
             connect_timeout: General::default_connect_timeout(),
             idle_timeout: General::default_idle_timeout(),
+            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(),
             tcp_user_timeout: Self::default_tcp_user_timeout(),
             log_client_connections: false,
             log_client_disconnections: false,
+            autoreload: None,
             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(),
-            idle_client_in_transaction_timeout: Self::default_idle_client_in_transaction_timeout(),
-            server_lifetime: Self::default_server_lifetime(),
-            server_round_robin: Self::default_server_round_robin(),
-            worker_threads: Self::default_worker_threads(),
-            autoreload: None,
             tls_certificate: None,
             tls_private_key: None,
             server_tls: false,
             verify_server_certificate: false,
             admin_username: String::from("admin"),
             admin_password: String::from("admin"),
-            admin_auth_type: AuthType::MD5,
-            validate_config: true,
             auth_query: None,
             auth_query_user: None,
             auth_query_password: None,
+            server_lifetime: Self::default_server_lifetime(),
+            server_round_robin: Self::default_server_round_robin(),
+            validate_config: true,
+            prepared_statements: false,
+            prepared_statements_cache_size: 500,
         }
     }
 }
@@ -492,20 +469,11 @@ pub enum PoolMode {
     Session,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy, Hash)]
+impl ToString for PoolMode {
-pub enum AuthType {
+    fn to_string(&self) -> String {
-    #[serde(alias = "trust", alias = "Trust")]
+        match *self {
-    Trust,
+            PoolMode::Transaction => "transaction".to_string(),
-
+            PoolMode::Session => "session".to_string(),
-    #[serde(alias = "md5", alias = "MD5")]
-    MD5,
-}
-
-impl std::fmt::Display for PoolMode {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            PoolMode::Transaction => write!(f, "transaction"),
-            PoolMode::Session => write!(f, "session"),
         }
     }
 }
@@ -518,13 +486,12 @@ pub enum LoadBalancingMode {
     #[serde(alias = "loc", alias = "LOC", alias = "least_outstanding_connections")]
     LeastOutstandingConnections,
 }
-
+impl ToString for LoadBalancingMode {
-impl std::fmt::Display for LoadBalancingMode {
+    fn to_string(&self) -> String {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match *self {
-        match self {
+            LoadBalancingMode::Random => "random".to_string(),
-            LoadBalancingMode::Random => write!(f, "random"),
             LoadBalancingMode::LeastOutstandingConnections => {
-                write!(f, "least_outstanding_connections")
+                "least_outstanding_connections".to_string()
             }
         }
     }
@@ -544,11 +511,6 @@ 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,
 
@@ -573,9 +535,6 @@ 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>,
@@ -583,12 +542,6 @@ pub struct Pool {
     #[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>,
@@ -614,10 +567,6 @@ impl Pool {
         PoolMode::Transaction
     }
 
-    pub fn default_default_shard() -> DefaultShard {
-        DefaultShard::default()
-    }
-
     pub fn default_load_balancing_mode() -> LoadBalancingMode {
         LoadBalancingMode::Random
     }
@@ -638,10 +587,6 @@ impl Pool {
         true
     }
 
-    pub fn default_prepared_statements_cache_size() -> usize {
-        0
-    }
-
     pub fn validate(&mut self) -> Result<(), Error> {
         match self.default_role.as_ref() {
             "any" => (),
@@ -682,25 +627,13 @@ 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
@@ -713,14 +646,7 @@ impl Pool {
             None => None,
         };
 
-        if let DefaultShard::Shard(shard_number) = self.default_shard {
+        for (_, user) in &self.users {
-            if shard_number >= self.shards.len() {
-                error!("Invalid shard {:?}", shard_number);
-                return Err(Error::BadConfig);
-            }
-        }
-
-        for user in self.users.values() {
             user.validate()?;
         }
 
@@ -733,29 +659,24 @@ 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,
-            connect_timeout: None,
-            idle_timeout: None,
-            server_lifetime: None,
             sharding_function: ShardingFunction::PgBigintHash,
             automatic_sharding_key: None,
+            connect_timeout: None,
+            idle_timeout: 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,
-            cleanup_server_connections: true,
+            server_lifetime: None,
-            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())]),
+            cleanup_server_connections: true,
-            users: BTreeMap::default(),
         }
     }
 }
@@ -767,50 +688,6 @@ 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,
@@ -867,13 +744,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"),
         }
     }
 }
@@ -886,26 +763,15 @@ 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: {}",
-            is_enabled(self.intercept.as_ref()),
+            self.intercept.is_some(),
-            is_enabled(self.table_access.as_ref()),
+            self.table_access.is_some(),
-            is_enabled(self.query_logger.as_ref()),
+            self.query_logger.is_some(),
-            is_enabled(self.prewarmer.as_ref()),
+            self.prewarmer.is_some(),
         )
     }
 }
@@ -916,47 +782,23 @@ 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() {
@@ -974,7 +816,6 @@ 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() {
@@ -1025,17 +866,15 @@ impl Config {
     pub fn fill_up_auth_query_config(&mut self) {
         for (_name, pool) in self.pools.iter_mut() {
             if pool.auth_query.is_none() {
-                pool.auth_query.clone_from(&self.general.auth_query);
+                pool.auth_query = self.general.auth_query.clone();
             }
 
             if pool.auth_query_user.is_none() {
-                pool.auth_query_user
+                pool.auth_query_user = self.general.auth_query_user.clone();
-                    .clone_from(&self.general.auth_query_user);
             }
 
             if pool.auth_query_password.is_none() {
-                pool.auth_query_password
+                pool.auth_query_password = self.general.auth_query_password.clone();
-                    .clone_from(&self.general.auth_query_password);
             }
         }
     }
@@ -1046,8 +885,8 @@ impl Default for Config {
         Config {
             path: Self::default_path(),
             general: General::default(),
-            plugins: None,
             pools: HashMap::default(),
+            plugins: None,
         }
     }
 }
@@ -1075,17 +914,6 @@ 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(),
@@ -1101,8 +929,8 @@ impl From<&Config> for std::collections::HashMap<String, String> {
                     (
                         format!("pools.{:?}.users", pool_name),
                         pool.users
-                            .values()
+                            .iter()
-                            .map(|user| &user.username)
+                            .map(|(_username, user)| &user.username)
                             .cloned()
                             .collect::<Vec<String>>()
                             .join(", "),
@@ -1156,7 +984,6 @@ 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",
@@ -1183,14 +1010,18 @@ impl Config {
             "Default max server lifetime: {}ms",
             self.general.server_lifetime
         );
-        info!("Server round robin: {}", self.general.server_round_robin);
+        info!("Sever round robin: {}", self.general.server_round_robin);
         match self.general.tls_certificate.clone() {
             Some(tls_certificate) => {
                 info!("TLS certificate: {}", tls_certificate);
 
-                if let Some(tls_private_key) = self.general.tls_private_key.clone() {
+                match self.general.tls_private_key.clone() {
-                    info!("TLS private key: {}", tls_private_key);
+                    Some(tls_private_key) => {
-                    info!("TLS support is enabled");
+                        info!("TLS private key: {}", tls_private_key);
+                        info!("TLS support is enabled");
+                    }
+
+                    None => (),
                 }
             }
 
@@ -1203,6 +1034,13 @@ impl Config {
             "Server TLS certificate verification: {}",
             self.general.verify_server_certificate
         );
+        info!("Prepared statements: {}", self.general.prepared_statements);
+        if self.general.prepared_statements {
+            info!(
+                "Prepared statements server cache size: {}",
+                self.general.prepared_statements_cache_size
+            );
+        }
         info!(
             "Plugins: {}",
             match self.plugins {
@@ -1218,8 +1056,8 @@ impl Config {
                 pool_name,
                 pool_config
                     .users
-                    .values()
+                    .iter()
-                    .map(|user_cfg| user_cfg.pool_size)
+                    .map(|(_, user_cfg)| user_cfg.pool_size)
                     .sum::<u32>()
                     .to_string()
             );
@@ -1258,15 +1096,6 @@ 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,
@@ -1289,14 +1118,6 @@ impl Config {
                 "[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,
@@ -1339,24 +1160,6 @@ 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(),
-                    }
-                );
             }
         }
     }
@@ -1411,31 +1214,34 @@ impl Config {
         }
 
         // Validate TLS!
-        if let Some(tls_certificate) = self.general.tls_certificate.clone() {
+        match self.general.tls_certificate.clone() {
-            match load_certs(Path::new(&tls_certificate)) {
+            Some(tls_certificate) => {
-                Ok(_) => {
+                match load_certs(Path::new(&tls_certificate)) {
-                    // Cert is okay, but what about the private key?
+                    Ok(_) => {
-                    match self.general.tls_private_key.clone() {
+                        // Cert is okay, but what about the private key?
-                        Some(tls_private_key) => match load_keys(Path::new(&tls_private_key)) {
+                        match self.general.tls_private_key.clone() {
-                            Ok(_) => (),
+                            Some(tls_private_key) => match load_keys(Path::new(&tls_private_key)) {
-                            Err(err) => {
+                                Ok(_) => (),
-                                error!("tls_private_key is incorrectly configured: {:?}", err);
+                                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");
                                 return Err(Error::BadConfig);
                             }
-                        },
+                        };
+                    }
 
-                        None => {
+                    Err(err) => {
-                            error!("tls_certificate is set, but the tls_private_key is not");
+                        error!("tls_certificate is incorrectly configured: {:?}", err);
-                            return Err(Error::BadConfig);
+                        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() {
@@ -1457,6 +1263,14 @@ pub fn get_idle_client_in_transaction_timeout() -> u64 {
     CONFIG.load().general.idle_client_in_transaction_timeout
 }
 
+pub fn get_prepared_statements() -> bool {
+    CONFIG.load().general.prepared_statements
+}
+
+pub fn get_prepared_statements_cache_size() -> usize {
+    CONFIG.load().general.prepared_statements_cache_size
+}
+
 /// Parse the configuration file located at the path.
 pub async fn parse(path: &str) -> Result<(), Error> {
     let mut contents = String::new();

src/errors.rs

@@ -12,7 +12,6 @@ pub enum Error {
     ProtocolSyncError(String),
     BadQuery(String),
     ServerError,
-    ServerMessageParserError(String),
     ServerStartupError(String, ServerIdentifier),
     ServerAuthError(String, ServerIdentifier),
     BadConfig,
@@ -28,8 +27,6 @@ pub enum Error {
     UnsupportedStatement,
     QueryRouterParserError(String),
     QueryRouterError(String),
-    InvalidShardId(usize),
-    PreparedStatementError,
 }
 
 #[derive(Clone, PartialEq, Debug)]

src/logger.rs

@@ -1,15 +1,9 @@
 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_max_level(args.log_level)
-        .with_env_filter(filter)
         .with_ansi(!args.no_color);
 
     match args.log_format {

src/main.rs

@@ -23,6 +23,7 @@ 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;

src/messages.rs

@@ -11,17 +11,11 @@ use crate::client::PREPARED_STATEMENT_COUNTER;
 use crate::config::get_config;
 use crate::errors::Error;
 
-use crate::constants::MESSAGE_TERMINATOR;
-use std::collections::hash_map::DefaultHasher;
 use std::collections::HashMap;
 use std::ffi::CString;
-use std::fmt::{Display, Formatter};
-use std::hash::{Hash, Hasher};
 use std::io::{BufRead, Cursor};
 use std::mem;
-use std::str::FromStr;
 use std::sync::atomic::Ordering;
-use std::sync::Arc;
 use std::time::Duration;
 
 /// Postgres data type mappings
@@ -117,11 +111,19 @@ pub fn simple_query(query: &str) -> BytesMut {
 }
 
 /// Tell the client we're ready for another query.
-pub async fn send_ready_for_query<S>(stream: &mut S) -> Result<(), Error>
+pub async fn ready_for_query<S>(stream: &mut S) -> Result<(), Error>
 where
     S: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    write_all(stream, ready_for_query(false)).await
+    let mut bytes = BytesMut::with_capacity(
+        mem::size_of::<u8>() + mem::size_of::<i32>() + mem::size_of::<u8>(),
+    );
+
+    bytes.put_u8(b'Z');
+    bytes.put_i32(5);
+    bytes.put_u8(b'I'); // Idle
+
+    write_all(stream, bytes).await
 }
 
 /// Send the startup packet the server. We're pretending we're a Pg client.
@@ -139,10 +141,6 @@ where
     bytes.put_slice(user.as_bytes());
     bytes.put_u8(0);
 
-    // Application name
-    bytes.put(&b"application_name\0"[..]);
-    bytes.put_slice(&b"pgcat\0"[..]);
-
     // Database
     bytes.put(&b"database\0"[..]);
     bytes.put_slice(database.as_bytes());
@@ -158,10 +156,12 @@ where
 
     match stream.write_all(&startup).await {
         Ok(_) => Ok(()),
-        Err(err) => Err(Error::SocketError(format!(
+        Err(err) => {
-            "Error writing startup to server socket - Error: {:?}",
+            return Err(Error::SocketError(format!(
-            err
+                "Error writing startup to server socket - Error: {:?}",
-        ))),
+                err
+            )))
+        }
     }
 }
 
@@ -237,8 +237,8 @@ pub fn md5_hash_password(user: &str, password: &str, salt: &[u8]) -> Vec<u8> {
     let mut md5 = Md5::new();
 
     // First pass
-    md5.update(password.as_bytes());
+    md5.update(&password.as_bytes());
-    md5.update(user.as_bytes());
+    md5.update(&user.as_bytes());
 
     let output = md5.finalize_reset();
 
@@ -274,7 +274,7 @@ where
 {
     let password = md5_hash_password(user, password, salt);
 
-    let mut message = BytesMut::with_capacity(password.len() + 5);
+    let mut message = BytesMut::with_capacity(password.len() as usize + 5);
 
     message.put_u8(b'p');
     message.put_i32(password.len() as i32 + 4);
@@ -288,7 +288,7 @@ where
     S: tokio::io::AsyncWrite + std::marker::Unpin,
 {
     let password = md5_hash_second_pass(hash, salt);
-    let mut message = BytesMut::with_capacity(password.len() + 5);
+    let mut message = BytesMut::with_capacity(password.len() as usize + 5);
 
     message.put_u8(b'p');
     message.put_i32(password.len() as i32 + 4);
@@ -315,7 +315,7 @@ where
     res.put_slice(&set_complete[..]);
 
     write_all_half(stream, &res).await?;
-    send_ready_for_query(stream).await
+    ready_for_query(stream).await
 }
 
 /// Send a custom error message to the client.
@@ -326,7 +326,7 @@ where
     S: tokio::io::AsyncWrite + std::marker::Unpin,
 {
     error_response_terminal(stream, message).await?;
-    send_ready_for_query(stream).await
+    ready_for_query(stream).await
 }
 
 /// Send a custom error message to the client.
@@ -427,7 +427,7 @@ where
     res.put(command_complete("SELECT 1"));
 
     write_all_half(stream, &res).await?;
-    send_ready_for_query(stream).await
+    ready_for_query(stream).await
 }
 
 pub fn row_description(columns: &Vec<(&str, DataType)>) -> BytesMut {
@@ -509,7 +509,7 @@ pub fn data_row_nullable(row: &Vec<Option<String>>) -> BytesMut {
             data_row.put_i32(column.len() as i32);
             data_row.put_slice(column);
         } else {
-            data_row.put_i32(-1_i32);
+            data_row.put_i32(-1 as i32);
         }
     }
 
@@ -557,37 +557,6 @@ pub fn flush() -> BytesMut {
     bytes
 }
 
-pub fn sync() -> BytesMut {
-    let mut bytes = BytesMut::with_capacity(mem::size_of::<u8>() + mem::size_of::<i32>());
-    bytes.put_u8(b'S');
-    bytes.put_i32(4);
-    bytes
-}
-
-pub fn parse_complete() -> BytesMut {
-    let mut bytes = BytesMut::with_capacity(mem::size_of::<u8>() + mem::size_of::<i32>());
-
-    bytes.put_u8(b'1');
-    bytes.put_i32(4);
-    bytes
-}
-
-pub fn ready_for_query(in_transaction: bool) -> BytesMut {
-    let mut bytes = BytesMut::with_capacity(
-        mem::size_of::<u8>() + mem::size_of::<i32>() + mem::size_of::<u8>(),
-    );
-
-    bytes.put_u8(b'Z');
-    bytes.put_i32(5);
-    if in_transaction {
-        bytes.put_u8(b'T');
-    } else {
-        bytes.put_u8(b'I');
-    }
-
-    bytes
-}
-
 /// Write all data in the buffer to the TcpStream.
 pub async fn write_all<S>(stream: &mut S, buf: BytesMut) -> Result<(), Error>
 where
@@ -595,10 +564,12 @@ where
 {
     match stream.write_all(&buf).await {
         Ok(_) => Ok(()),
-        Err(err) => Err(Error::SocketError(format!(
+        Err(err) => {
-            "Error writing to socket - Error: {:?}",
+            return Err(Error::SocketError(format!(
-            err
+                "Error writing to socket - Error: {:?}",
-        ))),
+                err
+            )))
+        }
     }
 }
 
@@ -609,10 +580,12 @@ where
 {
     match stream.write_all(buf).await {
         Ok(_) => Ok(()),
-        Err(err) => Err(Error::SocketError(format!(
+        Err(err) => {
-            "Error writing to socket - Error: {:?}",
+            return Err(Error::SocketError(format!(
-            err
+                "Error writing to socket - Error: {:?}",
-        ))),
+                err
+            )))
+        }
     }
 }
 
@@ -623,15 +596,19 @@ where
     match stream.write_all(buf).await {
         Ok(_) => match stream.flush().await {
             Ok(_) => Ok(()),
-            Err(err) => Err(Error::SocketError(format!(
+            Err(err) => {
-                "Error flushing socket - Error: {:?}",
+                return Err(Error::SocketError(format!(
-                err
+                    "Error flushing socket - Error: {:?}",
-            ))),
+                    err
+                )))
+            }
         },
-        Err(err) => Err(Error::SocketError(format!(
+        Err(err) => {
-            "Error writing to socket - Error: {:?}",
+            return Err(Error::SocketError(format!(
-            err
+                "Error writing to socket - Error: {:?}",
-        ))),
+                err
+            )))
+        }
     }
 }
 
@@ -733,10 +710,6 @@ pub fn configure_socket(stream: &TcpStream) {
         }
         Err(err) => error!("Could not configure socket: {}", err),
     }
-    match sock_ref.set_nodelay(true) {
-        Ok(_) => (),
-        Err(err) => error!("Could not configure TCP_NODELAY for socket: {}", err),
-    }
 }
 
 pub trait BytesMutReader {
@@ -750,71 +723,11 @@ impl BytesMutReader for Cursor<&BytesMut> {
         let mut buf = vec![];
         match self.read_until(b'\0', &mut buf) {
             Ok(_) => Ok(String::from_utf8_lossy(&buf[..buf.len() - 1]).to_string()),
-            Err(err) => Err(Error::ParseBytesError(err.to_string())),
+            Err(err) => return Err(Error::ParseBytesError(err.to_string())),
         }
     }
 }
 
-impl BytesMutReader for BytesMut {
-    /// Should only be used when reading strings from the message protocol.
-    /// Can be used to read multiple strings from the same message which are separated by the null byte
-    fn read_string(&mut self) -> Result<String, Error> {
-        let null_index = self.iter().position(|&byte| byte == b'\0');
-
-        match null_index {
-            Some(index) => {
-                let string_bytes = self.split_to(index + 1);
-                Ok(String::from_utf8_lossy(&string_bytes[..string_bytes.len() - 1]).to_string())
-            }
-            None => Err(Error::ParseBytesError("Could not read string".to_string())),
-        }
-    }
-}
-
-pub enum ExtendedProtocolData {
-    Parse {
-        data: BytesMut,
-        metadata: Option<(Arc<Parse>, u64)>,
-    },
-    Bind {
-        data: BytesMut,
-        metadata: Option<String>,
-    },
-    Describe {
-        data: BytesMut,
-        metadata: Option<String>,
-    },
-    Execute {
-        data: BytesMut,
-    },
-    Close {
-        data: BytesMut,
-        close: Close,
-    },
-}
-
-impl ExtendedProtocolData {
-    pub fn create_new_parse(data: BytesMut, metadata: Option<(Arc<Parse>, u64)>) -> Self {
-        Self::Parse { data, metadata }
-    }
-
-    pub fn create_new_bind(data: BytesMut, metadata: Option<String>) -> Self {
-        Self::Bind { data, metadata }
-    }
-
-    pub fn create_new_describe(data: BytesMut, metadata: Option<String>) -> Self {
-        Self::Describe { data, metadata }
-    }
-
-    pub fn create_new_execute(data: BytesMut) -> Self {
-        Self::Execute { data }
-    }
-
-    pub fn create_new_close(data: BytesMut, close: Close) -> Self {
-        Self::Close { data, close }
-    }
-}
-
 /// Parse (F) message.
 /// See: <https://www.postgresql.org/docs/current/protocol-message-formats.html>
 #[derive(Clone, Debug)]
@@ -823,6 +736,7 @@ pub struct Parse {
     #[allow(dead_code)]
     len: i32,
     pub name: String,
+    pub generated_name: String,
     query: String,
     num_params: i16,
     param_types: Vec<i32>,
@@ -848,6 +762,7 @@ impl TryFrom<&BytesMut> for Parse {
             code,
             len,
             name,
+            generated_name: prepared_statement_name(),
             query,
             num_params,
             param_types,
@@ -896,44 +811,11 @@ impl TryFrom<&Parse> for BytesMut {
 }
 
 impl Parse {
-    /// Renames the prepared statement to a new name based on the global counter
+    pub fn rename(mut self) -> Self {
-    pub fn rewrite(mut self) -> Self {
+        self.name = self.generated_name.to_string();
-        self.name = format!(
-            "PGCAT_{}",
-            PREPARED_STATEMENT_COUNTER.fetch_add(1, Ordering::SeqCst)
-        );
         self
     }
 
-    /// Gets the name of the prepared statement from the buffer
-    pub fn get_name(buf: &BytesMut) -> Result<String, Error> {
-        let mut cursor = Cursor::new(buf);
-        // Skip the code and length
-        cursor.advance(mem::size_of::<u8>() + mem::size_of::<i32>());
-        cursor.read_string()
-    }
-
-    /// Hashes the parse statement to be used as a key in the global cache
-    pub fn get_hash(&self) -> u64 {
-        // TODO_ZAIN: Take a look at which hashing function is being used
-        let mut hasher = DefaultHasher::new();
-
-        let concatenated = format!(
-            "{}{}{}",
-            self.query,
-            self.num_params,
-            self.param_types
-                .iter()
-                .map(ToString::to_string)
-                .collect::<Vec<_>>()
-                .join(",")
-        );
-
-        concatenated.hash(&mut hasher);
-
-        hasher.finish()
-    }
-
     pub fn anonymous(&self) -> bool {
         self.name.is_empty()
     }
@@ -1064,42 +946,9 @@ impl TryFrom<Bind> for BytesMut {
 }
 
 impl Bind {
-    /// Gets the name of the prepared statement from the buffer
+    pub fn reassign(mut self, parse: &Parse) -> Self {
-    pub fn get_name(buf: &BytesMut) -> Result<String, Error> {
+        self.prepared_statement = parse.name.clone();
-        let mut cursor = Cursor::new(buf);
+        self
-        // Skip the code and length
-        cursor.advance(mem::size_of::<u8>() + mem::size_of::<i32>());
-        cursor.read_string()?;
-        cursor.read_string()
-    }
-
-    /// Renames the prepared statement to a new name
-    pub fn rename(buf: BytesMut, new_name: &str) -> Result<BytesMut, Error> {
-        let mut cursor = Cursor::new(&buf);
-        // Read basic data from the cursor
-        let code = cursor.get_u8();
-        let current_len = cursor.get_i32();
-        let portal = cursor.read_string()?;
-        let prepared_statement = cursor.read_string()?;
-
-        // Calculate new length
-        let new_len = current_len + new_name.len() as i32 - prepared_statement.len() as i32;
-
-        // Begin building the response buffer
-        let mut response_buf = BytesMut::with_capacity(new_len as usize + 1);
-        response_buf.put_u8(code);
-        response_buf.put_i32(new_len);
-
-        // Put the portal and new name into the buffer
-        // Note: panic if the provided string contains null byte
-        response_buf.put_slice(CString::new(portal)?.as_bytes_with_nul());
-        response_buf.put_slice(CString::new(new_name)?.as_bytes_with_nul());
-
-        // Add the remainder of the original buffer into the response
-        response_buf.put_slice(&buf[cursor.position() as usize..]);
-
-        // Return the buffer
-        Ok(response_buf)
     }
 
     pub fn anonymous(&self) -> bool {
@@ -1113,7 +962,7 @@ pub struct Describe {
 
     #[allow(dead_code)]
     len: i32,
-    pub target: char,
+    target: char,
     pub statement_name: String,
 }
 
@@ -1155,15 +1004,6 @@ impl TryFrom<Describe> for BytesMut {
 }
 
 impl Describe {
-    pub fn empty_new() -> Describe {
-        Describe {
-            code: 'D',
-            len: 4 + 1 + 1,
-            target: 'S',
-            statement_name: "".to_string(),
-        }
-    }
-
     pub fn rename(mut self, name: &str) -> Self {
         self.statement_name = name.to_string();
         self
@@ -1252,297 +1092,9 @@ pub fn close_complete() -> BytesMut {
     bytes
 }
 
-// from https://www.postgresql.org/docs/12/protocol-error-fields.html
+pub fn prepared_statement_name() -> String {
-#[derive(Debug, Default, PartialEq)]
+    format!(
-pub struct PgErrorMsg {
+        "P_{}",
-    pub severity_localized: String,      // S
+        PREPARED_STATEMENT_COUNTER.fetch_add(1, Ordering::SeqCst)
-    pub severity: String,                // V
+    )
-    pub code: String,                    // C
-    pub message: String,                 // M
-    pub detail: Option<String>,          // D
-    pub hint: Option<String>,            // H
-    pub position: Option<u32>,           // P
-    pub internal_position: Option<u32>,  // p
-    pub internal_query: Option<String>,  // q
-    pub where_context: Option<String>,   // W
-    pub schema_name: Option<String>,     // s
-    pub table_name: Option<String>,      // t
-    pub column_name: Option<String>,     // c
-    pub data_type_name: Option<String>,  // d
-    pub constraint_name: Option<String>, // n
-    pub file_name: Option<String>,       // F
-    pub line: Option<u32>,               // L
-    pub routine: Option<String>,         // R
-}
-
-// TODO: implement with https://docs.rs/derive_more/latest/derive_more/
-impl Display for PgErrorMsg {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "[severity: {}]", self.severity)?;
-        write!(f, "[code: {}]", self.code)?;
-        write!(f, "[message: {}]", self.message)?;
-        if let Some(val) = &self.detail {
-            write!(f, "[detail: {val}]")?;
-        }
-        if let Some(val) = &self.hint {
-            write!(f, "[hint: {val}]")?;
-        }
-        if let Some(val) = &self.position {
-            write!(f, "[position: {val}]")?;
-        }
-        if let Some(val) = &self.internal_position {
-            write!(f, "[internal_position: {val}]")?;
-        }
-        if let Some(val) = &self.internal_query {
-            write!(f, "[internal_query: {val}]")?;
-        }
-        if let Some(val) = &self.internal_query {
-            write!(f, "[internal_query: {val}]")?;
-        }
-        if let Some(val) = &self.where_context {
-            write!(f, "[where: {val}]")?;
-        }
-        if let Some(val) = &self.schema_name {
-            write!(f, "[schema_name: {val}]")?;
-        }
-        if let Some(val) = &self.table_name {
-            write!(f, "[table_name: {val}]")?;
-        }
-        if let Some(val) = &self.column_name {
-            write!(f, "[column_name: {val}]")?;
-        }
-        if let Some(val) = &self.data_type_name {
-            write!(f, "[data_type_name: {val}]")?;
-        }
-        if let Some(val) = &self.constraint_name {
-            write!(f, "[constraint_name: {val}]")?;
-        }
-        if let Some(val) = &self.file_name {
-            write!(f, "[file_name: {val}]")?;
-        }
-        if let Some(val) = &self.line {
-            write!(f, "[line: {val}]")?;
-        }
-        if let Some(val) = &self.routine {
-            write!(f, "[routine: {val}]")?;
-        }
-
-        write!(f, " ")?;
-
-        Ok(())
-    }
-}
-
-impl PgErrorMsg {
-    pub fn parse(error_msg: &[u8]) -> Result<PgErrorMsg, Error> {
-        let mut out = PgErrorMsg {
-            severity_localized: "".to_string(),
-            severity: "".to_string(),
-            code: "".to_string(),
-            message: "".to_string(),
-            detail: None,
-            hint: None,
-            position: None,
-            internal_position: None,
-            internal_query: None,
-            where_context: None,
-            schema_name: None,
-            table_name: None,
-            column_name: None,
-            data_type_name: None,
-            constraint_name: None,
-            file_name: None,
-            line: None,
-            routine: None,
-        };
-        for msg_part in error_msg.split(|v| *v == MESSAGE_TERMINATOR) {
-            if msg_part.is_empty() {
-                continue;
-            }
-
-            let msg_content = match String::from_utf8_lossy(&msg_part[1..]).parse() {
-                Ok(c) => c,
-                Err(err) => {
-                    return Err(Error::ServerMessageParserError(format!(
-                        "could not parse server message field. err {:?}",
-                        err
-                    )))
-                }
-            };
-
-            match &msg_part[0] {
-                b'S' => {
-                    out.severity_localized = msg_content;
-                }
-                b'V' => {
-                    out.severity = msg_content;
-                }
-                b'C' => {
-                    out.code = msg_content;
-                }
-                b'M' => {
-                    out.message = msg_content;
-                }
-                b'D' => {
-                    out.detail = Some(msg_content);
-                }
-                b'H' => {
-                    out.hint = Some(msg_content);
-                }
-                b'P' => out.position = Some(u32::from_str(msg_content.as_str()).unwrap_or(0)),
-                b'p' => {
-                    out.internal_position = Some(u32::from_str(msg_content.as_str()).unwrap_or(0))
-                }
-                b'q' => {
-                    out.internal_query = Some(msg_content);
-                }
-                b'W' => {
-                    out.where_context = Some(msg_content);
-                }
-                b's' => {
-                    out.schema_name = Some(msg_content);
-                }
-                b't' => {
-                    out.table_name = Some(msg_content);
-                }
-                b'c' => {
-                    out.column_name = Some(msg_content);
-                }
-                b'd' => {
-                    out.data_type_name = Some(msg_content);
-                }
-                b'n' => {
-                    out.constraint_name = Some(msg_content);
-                }
-                b'F' => {
-                    out.file_name = Some(msg_content);
-                }
-                b'L' => out.line = Some(u32::from_str(msg_content.as_str()).unwrap_or(0)),
-                b'R' => {
-                    out.routine = Some(msg_content);
-                }
-                _ => {}
-            }
-        }
-
-        Ok(out)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use crate::messages::PgErrorMsg;
-    use log::{error, info};
-
-    fn field(kind: char, content: &str) -> Vec<u8> {
-        format!("{kind}{content}\0").as_bytes().to_vec()
-    }
-
-    #[test]
-    fn parse_fields() {
-        let mut complete_msg = vec![];
-        let severity = "FATAL";
-        complete_msg.extend(field('S', severity));
-        complete_msg.extend(field('V', severity));
-
-        let error_code = "29P02";
-        complete_msg.extend(field('C', error_code));
-        let message = "password authentication failed for user \"wrong_user\"";
-        complete_msg.extend(field('M', message));
-        let detail_msg = "super detailed message";
-        complete_msg.extend(field('D', detail_msg));
-        let hint_msg = "hint detail here";
-        complete_msg.extend(field('H', hint_msg));
-        complete_msg.extend(field('P', "123"));
-        complete_msg.extend(field('p', "234"));
-        let internal_query = "SELECT * from foo;";
-        complete_msg.extend(field('q', internal_query));
-        let where_msg = "where goes here";
-        complete_msg.extend(field('W', where_msg));
-        let schema_msg = "schema_name";
-        complete_msg.extend(field('s', schema_msg));
-        let table_msg = "table_name";
-        complete_msg.extend(field('t', table_msg));
-        let column_msg = "column_name";
-        complete_msg.extend(field('c', column_msg));
-        let data_type_msg = "type_name";
-        complete_msg.extend(field('d', data_type_msg));
-        let constraint_msg = "constraint_name";
-        complete_msg.extend(field('n', constraint_msg));
-        let file_msg = "pgcat.c";
-        complete_msg.extend(field('F', file_msg));
-        complete_msg.extend(field('L', "335"));
-        let routine_msg = "my_failing_routine";
-        complete_msg.extend(field('R', routine_msg));
-
-        tracing_subscriber::fmt()
-            .with_max_level(tracing::Level::INFO)
-            .with_ansi(true)
-            .init();
-
-        info!(
-            "full message: {}",
-            PgErrorMsg::parse(&complete_msg).unwrap()
-        );
-        assert_eq!(
-            PgErrorMsg {
-                severity_localized: severity.to_string(),
-                severity: severity.to_string(),
-                code: error_code.to_string(),
-                message: message.to_string(),
-                detail: Some(detail_msg.to_string()),
-                hint: Some(hint_msg.to_string()),
-                position: Some(123),
-                internal_position: Some(234),
-                internal_query: Some(internal_query.to_string()),
-                where_context: Some(where_msg.to_string()),
-                schema_name: Some(schema_msg.to_string()),
-                table_name: Some(table_msg.to_string()),
-                column_name: Some(column_msg.to_string()),
-                data_type_name: Some(data_type_msg.to_string()),
-                constraint_name: Some(constraint_msg.to_string()),
-                file_name: Some(file_msg.to_string()),
-                line: Some(335),
-                routine: Some(routine_msg.to_string()),
-            },
-            PgErrorMsg::parse(&complete_msg).unwrap()
-        );
-
-        let mut only_mandatory_msg = vec![];
-        only_mandatory_msg.extend(field('S', severity));
-        only_mandatory_msg.extend(field('V', severity));
-        only_mandatory_msg.extend(field('C', error_code));
-        only_mandatory_msg.extend(field('M', message));
-        only_mandatory_msg.extend(field('D', detail_msg));
-
-        let err_fields = PgErrorMsg::parse(&only_mandatory_msg).unwrap();
-        info!("only mandatory fields: {}", &err_fields);
-        error!(
-            "server error: {}: {}",
-            err_fields.severity, err_fields.message
-        );
-        assert_eq!(
-            PgErrorMsg {
-                severity_localized: severity.to_string(),
-                severity: severity.to_string(),
-                code: error_code.to_string(),
-                message: message.to_string(),
-                detail: Some(detail_msg.to_string()),
-                hint: None,
-                position: None,
-                internal_position: None,
-                internal_query: None,
-                where_context: None,
-                schema_name: None,
-                table_name: None,
-                column_name: None,
-                data_type_name: None,
-                constraint_name: None,
-                file_name: None,
-                line: None,
-                routine: None,
-            },
-            PgErrorMsg::parse(&only_mandatory_msg).unwrap()
-        );
-    }
 }

src/mirrors.rs

@@ -23,15 +23,14 @@ 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, prepared_statement_cache_size) =
+        let (connection_timeout, idle_timeout, _cfg) =
             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(), 0),
+                None => (default, default, crate::config::Pool::default()),
             };
 
         let manager = ServerPool::new(
@@ -42,8 +41,6 @@ impl MirroredClient {
             Arc::new(RwLock::new(None)),
             None,
             true,
-            false,
-            prepared_statement_cache_size,
         );
 
         Pool::builder()
@@ -81,13 +78,12 @@ impl MirroredClient {
                     }
 
                     // Incoming data from server (we read to clear the socket buffer and discard the data)
-                    recv_result = server.recv(None) => {
+                    recv_result = server.recv() => {
                         match recv_result {
                             Ok(message) => trace!("Received from mirror: {} {:?}", String::from_utf8_lossy(&message[..]), address.clone()),
                             Err(err) => {
-                                server.mark_bad(
+                                server.mark_bad();
-                                    format!("Failed to send to mirror, Discarding message {:?}, {:?}", err, address.clone()).as_str()
+                                error!("Failed to receive from mirror {:?} {:?}", err, address.clone());
-                                );
                             }
                         }
                     }
@@ -99,9 +95,8 @@ impl MirroredClient {
                                 match server.send(&BytesMut::from(&bytes[..])).await {
                                     Ok(_) => trace!("Sent to mirror: {} {:?}", String::from_utf8_lossy(&bytes[..]), address.clone()),
                                     Err(err) => {
-                                        server.mark_bad(
+                                        server.mark_bad();
-                                            format!("Failed to receive from mirror {:?} {:?}", err, address.clone()).as_str()
+                                        error!("Failed to send to mirror, Discarding message {:?}, {:?}", err, address.clone())
-                                        );
                                     }
                                 }
                             }
@@ -141,18 +136,18 @@ impl MirroringManager {
                 bytes_rx,
                 disconnect_rx: exit_rx,
             };
-            exit_senders.push(exit_tx);
+            exit_senders.push(exit_tx.clone());
-            byte_senders.push(bytes_tx);
+            byte_senders.push(bytes_tx.clone());
             client.start();
         });
 
         Self {
-            byte_senders,
+            byte_senders: byte_senders,
             disconnect_senders: exit_senders,
         }
     }
 
-    pub fn send(&mut self, bytes: &BytesMut) {
+    pub fn send(self: &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
@@ -174,7 +169,7 @@ impl MirroringManager {
         });
     }
 
-    pub fn disconnect(&mut self) {
+    pub fn disconnect(self: &mut Self) {
         self.disconnect_senders
             .iter_mut()
             .for_each(|sender| match sender.try_send(()) {

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.is_empty() {
+                                if s == "" {
                                     None
                                 } else {
                                     Some(s)

src/plugins/mod.rs

@@ -33,7 +33,6 @@ 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.db, self.user, query);
+        info!("[pool: {}][user: {}] {}", self.user, self.db, 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,9 +1,9 @@
 use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use bb8::{ManageConnection, Pool, PooledConnection, QueueStrategy};
+use bytes::{BufMut, BytesMut};
 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;
@@ -11,8 +11,6 @@ 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,
@@ -21,14 +19,13 @@ use std::time::Instant;
 use tokio::sync::Notify;
 
 use crate::config::{
-    get_config, Address, DefaultShard, General, LoadBalancingMode, Plugins, PoolMode, Role, User,
+    get_config, Address, 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, ServerParameters};
+use crate::server::Server;
 use crate::sharding::ShardingFunction;
 use crate::stats::{AddressStats, ClientStats, ServerStats};
 
@@ -57,57 +54,6 @@ 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)]
@@ -165,12 +111,6 @@ 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,
 
@@ -195,9 +135,6 @@ 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,
 
@@ -220,8 +157,6 @@ 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,
@@ -231,7 +166,6 @@ 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,
@@ -244,23 +178,23 @@ impl Default for PoolSettings {
 #[derive(Clone, Debug, Default)]
 pub struct ConnectionPool {
     /// The pools handled internally by bb8.
-    databases: Arc<Vec<Vec<Pool<ServerPool>>>>,
+    databases: Vec<Vec<Pool<ServerPool>>>,
 
     /// The addresses (host, port, role) to handle
     /// failover and load balancing deterministically.
-    addresses: Arc<Vec<Vec<Address>>>,
+    addresses: Vec<Vec<Address>>,
 
     /// List of banned addresses (see above)
     /// that should not be queried.
     banlist: BanList,
 
-    /// The server information has to be passed to the
+    /// The server information (K messages) have to be passed to the
     /// clients on startup. We pre-connect to all shards and replicas
-    /// on pool creation and save the startup parameters here.
+    /// on pool creation and save the K messages here.
-    original_server_parameters: Arc<RwLock<ServerParameters>>,
+    server_info: Arc<RwLock<BytesMut>>,
 
     /// Pool configuration.
-    pub settings: Arc<PoolSettings>,
+    pub settings: PoolSettings,
 
     /// If not validated, we need to double check the pool is available before allowing a client
     /// to use it.
@@ -277,9 +211,6 @@ pub struct ConnectionPool {
 
     /// AuthInfo
     pub auth_hash: Arc<RwLock<Option<String>>>,
-
-    /// Cache
-    pub prepared_statement_cache: Option<PreparedStatementCacheType>,
 }
 
 impl ConnectionPool {
@@ -298,17 +229,20 @@ impl ConnectionPool {
                 let old_pool_ref = get_pool(pool_name, &user.username);
                 let identifier = PoolIdentifier::new(pool_name, &user.username);
 
-                if let Some(pool) = old_pool_ref {
+                match old_pool_ref {
-                    // If the pool hasn't changed, get existing reference and insert it into the new_pools.
+                    Some(pool) => {
-                    // We replace all pools at the end, but if the reference is kept, the pool won't get re-created (bb8).
+                        // If the pool hasn't changed, get existing reference and insert it into the new_pools.
-                    if pool.config_hash == new_pool_hash_value {
+                        // We replace all pools at the end, but if the reference is kept, the pool won't get re-created (bb8).
-                        info!(
+                        if pool.config_hash == new_pool_hash_value {
-                            "[pool: {}][user: {}] has not changed",
+                            info!(
-                            pool_name, user.username
+                                "[pool: {}][user: {}] has not changed",
-                        );
+                                pool_name, user.username
-                        new_pools.insert(identifier.clone(), pool.clone());
+                            );
-                        continue;
+                            new_pools.insert(identifier.clone(), pool.clone());
+                            continue;
+                        }
                     }
+                    None => (),
                 }
 
                 info!(
@@ -358,7 +292,6 @@ impl ConnectionPool {
                                     pool_name: pool_name.clone(),
                                     mirrors: vec![],
                                     stats: Arc::new(AddressStats::default()),
-                                    error_count: Arc::new(AtomicU64::new(0)),
                                 });
                                 address_id += 1;
                             }
@@ -377,7 +310,6 @@ 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;
@@ -432,24 +364,16 @@ impl ConnectionPool {
                                 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 user.connect_timeout {
+                        let connect_timeout = match pool_config.connect_timeout {
                             Some(connect_timeout) => connect_timeout,
-                            None => match pool_config.connect_timeout {
+                            None => config.general.connect_timeout,
-                                Some(connect_timeout) => connect_timeout,
-                                None => config.general.connect_timeout,
-                            },
                         };
 
-                        let idle_timeout = match user.idle_timeout {
+                        let idle_timeout = match pool_config.idle_timeout {
                             Some(idle_timeout) => idle_timeout,
-                            None => match pool_config.idle_timeout {
+                            None => config.general.idle_timeout,
-                                Some(idle_timeout) => idle_timeout,
-                                None => config.general.idle_timeout,
-                            },
                         };
 
                         let server_lifetime = match user.server_lifetime {
@@ -460,7 +384,7 @@ impl ConnectionPool {
                             },
                         };
 
-                        let reaper_rate = *[idle_timeout, server_lifetime, POOL_REAPER_RATE]
+                        let reaper_rate = *vec![idle_timeout, server_lifetime, POOL_REAPER_RATE]
                             .iter()
                             .min()
                             .unwrap();
@@ -509,13 +433,13 @@ impl ConnectionPool {
                 }
 
                 let pool = ConnectionPool {
-                    databases: Arc::new(shards),
+                    databases: shards,
-                    addresses: Arc::new(addresses),
+                    addresses,
                     banlist: Arc::new(RwLock::new(banlist)),
                     config_hash: new_pool_hash_value,
-                    original_server_parameters: Arc::new(RwLock::new(ServerParameters::new())),
+                    server_info: Arc::new(RwLock::new(BytesMut::new())),
                     auth_hash: pool_auth_hash,
-                    settings: Arc::new(PoolSettings {
+                    settings: PoolSettings {
                         pool_mode: match user.pool_mode {
                             Some(pool_mode) => pool_mode,
                             None => pool_config.pool_mode,
@@ -532,9 +456,6 @@ 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(),
@@ -550,7 +471,6 @@ 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(),
@@ -558,23 +478,17 @@ 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 validate_pool = pool.clone();
+                    let mut validate_pool = pool.clone();
                     tokio::task::spawn(async move {
                         let _ = validate_pool.validate().await;
                     });
@@ -595,7 +509,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(&self) -> Result<(), Error> {
+    pub async fn validate(&mut self) -> Result<(), Error> {
         let mut futures = Vec::new();
         let validated = Arc::clone(&self.validated);
 
@@ -603,7 +517,7 @@ impl ConnectionPool {
             for server in 0..self.servers(shard) {
                 let databases = self.databases.clone();
                 let validated = Arc::clone(&validated);
-                let pool_server_parameters = Arc::clone(&self.original_server_parameters);
+                let pool_server_info = Arc::clone(&self.server_info);
 
                 let task = tokio::task::spawn(async move {
                     let connection = match databases[shard][server].get().await {
@@ -616,10 +530,11 @@ impl ConnectionPool {
 
                     let proxy = connection;
                     let server = &*proxy;
-                    let server_parameters: ServerParameters = server.server_parameters();
+                    let server_info = server.server_info();
 
-                    let mut guard = pool_server_parameters.write();
+                    let mut guard = pool_server_info.write();
-                    *guard = server_parameters;
+                    guard.clear();
+                    guard.put(server_info.clone());
                     validated.store(true, Ordering::Relaxed);
                 });
 
@@ -631,7 +546,7 @@ impl ConnectionPool {
 
         // TODO: compare server information to make sure
         // all shards are running identical configurations.
-        if !self.validated() {
+        if self.server_info.read().is_empty() {
             error!("Could not validate connection pool");
             return Err(Error::AllServersDown);
         }
@@ -678,51 +593,19 @@ impl ConnectionPool {
     /// Get a connection from the pool.
     pub async fn get(
         &self,
-        shard: Option<usize>,       // shard number
+        shard: usize,               // shard number
         role: Option<Role>,         // primary or replica
         client_stats: &ClientStats, // client id
     ) -> Result<(PooledConnection<'_, ServerPool>, Address), Error> {
-        let effective_shard_id = if self.shards() == 1 {
+        let mut candidates: Vec<&Address> = self.addresses[shard]
-            // 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
-        };
-
-        let mut candidates = self
-            .addresses
             .iter()
-            .flatten()
             .filter(|address| address.role == role)
-            .collect::<Vec<&Address>>();
+            .collect();
 
-        // We start with a shuffled list of addresses even if we end up resorting
+        // We shuffle even if least_outstanding_queries is used to avoid imbalance
-        // this is meant to avoid hitting instance 0 everytime if the sorting metric
+        // in cases where all candidates have more or less the same number of outstanding
-        // ends up being the same for all instances
+        // queries
         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)
@@ -745,7 +628,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);
@@ -758,10 +641,7 @@ impl ConnectionPool {
                 .get()
                 .await
             {
-                Ok(conn) => {
+                Ok(conn) => conn,
-                    address.reset_error_count();
-                    conn
-                }
                 Err(err) => {
                     error!(
                         "Connection checkout error for instance {:?}, error: {:?}",
@@ -769,6 +649,7 @@ impl ConnectionPool {
                     );
                     self.ban(address, BanReason::FailedCheckout, Some(client_stats));
                     address.stats.error();
+                    client_stats.idle();
                     client_stats.checkout_error();
                     continue;
                 }
@@ -786,8 +667,8 @@ impl ConnectionPool {
             // since we last checked the server is ok.
             // Health checks are pretty expensive.
             if !require_healthcheck {
-                let checkout_time = now.elapsed().as_micros() as u64;
+                let checkout_time: u64 = now.elapsed().as_micros() as u64;
-                client_stats.checkout_success();
+                client_stats.checkout_time(checkout_time);
                 server
                     .stats()
                     .checkout_time(checkout_time, client_stats.application_name());
@@ -800,8 +681,8 @@ impl ConnectionPool {
                 .run_health_check(address, server, now, client_stats)
                 .await
             {
-                let checkout_time = now.elapsed().as_micros() as u64;
+                let checkout_time: u64 = now.elapsed().as_micros() as u64;
-                client_stats.checkout_success();
+                client_stats.checkout_time(checkout_time);
                 server
                     .stats()
                     .checkout_time(checkout_time, client_stats.application_name());
@@ -812,9 +693,7 @@ impl ConnectionPool {
                 continue;
             }
         }
-
+        client_stats.idle();
-        client_stats.checkout_error();
-
         Err(Error::AllServersDown)
     }
 
@@ -839,7 +718,7 @@ impl ConnectionPool {
             Ok(res) => match res {
                 Ok(_) => {
                     let checkout_time: u64 = start.elapsed().as_micros() as u64;
-                    client_info.checkout_success();
+                    client_info.checkout_time(checkout_time);
                     server
                         .stats()
                         .checkout_time(checkout_time, client_info.application_name());
@@ -867,28 +746,16 @@ impl ConnectionPool {
         }
 
         // Don't leave a bad connection in the pool.
-        server.mark_bad("failed health check");
+        server.mark_bad();
 
-        self.ban(address, BanReason::FailedHealthCheck, Some(client_info));
+        self.ban(&address, BanReason::FailedHealthCheck, Some(client_info));
-        false
+        return 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;
@@ -994,10 +861,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)));
+                bans.push((address.clone(), (reason.clone(), timestamp.clone())));
             }
         }
-        bans
+        return bans;
     }
 
     /// Get the address from the host url
@@ -1039,11 +906,10 @@ impl ConnectionPool {
         &self.addresses[shard][server]
     }
 
-    pub fn server_parameters(&self) -> ServerParameters {
+    pub fn server_info(&self) -> BytesMut {
-        self.original_server_parameters.read().clone()
+        self.server_info.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;
@@ -1055,37 +921,7 @@ impl ConnectionPool {
         }
         let busy = provisioned - idle;
         debug!("{:?} has {:?} busy connections", address, busy);
-        busy
+        return 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);
-        }
     }
 }
 
@@ -1111,16 +947,9 @@ pub struct ServerPool {
 
     /// 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,
@@ -1129,19 +958,15 @@ impl ServerPool {
         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: user.clone(),
             database: database.to_string(),
             client_server_map,
             auth_hash,
             plugins,
             cleanup_connections,
-            log_client_parameter_status_changes,
-            prepared_statement_cache_size,
         }
     }
 }
@@ -1171,8 +996,6 @@ impl ManageConnection for ServerPool {
             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,41 +1,23 @@
-use http_body_util::Full;
+use hyper::service::{make_service_fn, service_fn};
-use hyper::body;
+use hyper::{Body, Method, Request, Response, Server, StatusCode};
-use hyper::body::Bytes;
-
-use hyper::server::conn::http1;
-use hyper::service::service_fn;
-use hyper::{Method, Request, Response, StatusCode};
-use hyper_util::rt::TokioIo;
 use log::{debug, error, info};
 use phf::phf_map;
 use std::collections::HashMap;
 use std::fmt;
 use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
-use tokio::net::TcpListener;
+use std::sync::Arc;
 
 use crate::config::Address;
 use crate::pool::{get_all_pools, PoolIdentifier};
-use crate::stats::get_server_stats;
 use crate::stats::pool::PoolStats;
+use crate::stats::{get_server_stats, ServerStats};
 
 struct MetricHelpType {
     help: &'static str,
     ty: &'static str,
 }
 
-struct ServerPrometheusStats {
-    bytes_received: u64,
-    bytes_sent: u64,
-    transaction_count: u64,
-    query_count: u64,
-    error_count: u64,
-    active_count: u64,
-    idle_count: u64,
-    login_count: u64,
-    tested_count: u64,
-}
-
 // reference for metric types: https://prometheus.io/docs/concepts/metric_types/
 // counters only increase
 // gauges can arbitrarily increase or decrease
@@ -138,46 +120,22 @@ static METRIC_HELP_AND_TYPES_LOOKUP: phf::Map<&'static str, MetricHelpType> = ph
     },
     "servers_bytes_received" => MetricHelpType {
         help: "Volume in bytes of network traffic received by server",
-        ty: "counter",
+        ty: "gauge",
     },
     "servers_bytes_sent" => MetricHelpType {
         help: "Volume in bytes of network traffic sent by server",
-        ty: "counter",
+        ty: "gauge",
     },
     "servers_transaction_count" => MetricHelpType {
         help: "Number of transactions executed by server",
-        ty: "counter",
+        ty: "gauge",
     },
     "servers_query_count" => MetricHelpType {
         help: "Number of queries executed by server",
-        ty: "counter",
+        ty: "gauge",
     },
     "servers_error_count" => MetricHelpType {
         help: "Number of errors",
-        ty: "counter",
-    },
-    "servers_idle_count" => MetricHelpType {
-        help: "Number of server connection in idle state",
-        ty: "gauge",
-    },
-    "servers_active_count" => MetricHelpType {
-        help: "Number of server connection in active state",
-        ty: "gauge",
-    },
-    "servers_tested_count" => MetricHelpType {
-        help: "Number of server connection in tested state",
-        ty: "gauge",
-    },
-    "servers_login_count" => MetricHelpType {
-        help: "Number of server connection in login state",
-        ty: "gauge",
-    },
-    "servers_is_banned" => MetricHelpType {
-        help: "0 if server is not banned, 1 if server is banned",
-        ty: "gauge",
-    },
-    "servers_is_paused" => MetricHelpType {
-        help: "0 if server is not paused, 1 if server is paused",
         ty: "gauge",
     },
     "databases_pool_size" => MetricHelpType {
@@ -200,17 +158,18 @@ struct PrometheusMetric<Value: fmt::Display> {
 
 impl<Value: fmt::Display> fmt::Display for PrometheusMetric<Value> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let mut sorted_labels: Vec<_> = self.labels.iter().collect();
+        let formatted_labels = self
-        sorted_labels.sort_by_key(|&(key, _)| key);
+            .labels
-        let formatted_labels = sorted_labels
             .iter()
             .map(|(key, value)| format!("{}=\"{}\"", key, value))
             .collect::<Vec<_>>()
             .join(",");
         write!(
             f,
-            "{name}{{{formatted_labels}}} {value}",
+            "# HELP {name} {help}\n# TYPE {name} {ty}\n{name}{{{formatted_labels}}} {value}\n",
             name = format_args!("pgcat_{}", self.name),
+            help = self.help,
+            ty = self.ty,
             formatted_labels = formatted_labels,
             value = self.value
         )
@@ -244,9 +203,7 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("shard", address.shard.to_string());
         labels.insert("role", address.role.to_string());
         labels.insert("pool", address.pool_name.clone());
-        labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("username", address.username.clone());
 
         Self::from_name(&format!("databases_{}", name), value, labels)
     }
@@ -261,9 +218,7 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("shard", address.shard.to_string());
         labels.insert("role", address.role.to_string());
         labels.insert("pool", address.pool_name.clone());
-        labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("username", address.username.clone());
 
         Self::from_name(&format!("servers_{}", name), value, labels)
     }
@@ -274,9 +229,7 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("shard", address.shard.to_string());
         labels.insert("pool", address.pool_name.clone());
         labels.insert("role", address.role.to_string());
-        labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("username", address.username.clone());
 
         Self::from_name(&format!("stats_{}", name), value, labels)
     }
@@ -288,20 +241,9 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
 
         Self::from_name(&format!("pools_{}", name), value, labels)
     }
-
-    fn get_header(&self) -> String {
-        format!(
-            "\n# HELP {name} {help}\n# TYPE {name} {ty}",
-            name = format_args!("pgcat_{}", self.name),
-            help = self.help,
-            ty = self.ty,
-        )
-    }
 }
 
-async fn prometheus_stats(
+async fn prometheus_stats(request: Request<Body>) -> Result<Response<Body>, hyper::http::Error> {
-    request: Request<body::Incoming>,
-) -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     match (request.method(), request.uri().path()) {
         (&Method::GET, "/metrics") => {
             let mut lines = Vec::new();
@@ -309,7 +251,6 @@ async fn prometheus_stats(
             push_pool_stats(&mut lines);
             push_server_stats(&mut lines);
             push_database_stats(&mut lines);
-            lines.push("".to_string()); // Ensure to end the stats with a line terminator as required by the specification.
 
             Response::builder()
                 .header("content-type", "text/plain; version=0.0.4")
@@ -323,7 +264,6 @@ async fn prometheus_stats(
 
 // Adds metrics shown in a SHOW STATS admin command.
 fn push_address_stats(lines: &mut Vec<String>) {
-    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
     for (_, pool) in get_all_pools() {
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
@@ -333,10 +273,7 @@ fn push_address_stats(lines: &mut Vec<String>) {
                     if let Some(prometheus_metric) =
                         PrometheusMetric::<u64>::from_address(address, &key, value)
                     {
-                        grouped_metrics
+                        lines.push(prometheus_metric.to_string());
-                            .entry(key)
-                            .or_default()
-                            .push(prometheus_metric);
                     } else {
                         debug!("Metric {} not implemented for {}", key, address.name());
                     }
@@ -344,53 +281,33 @@ fn push_address_stats(lines: &mut Vec<String>) {
             }
         }
     }
-    for (_key, metrics) in grouped_metrics {
-        if !metrics.is_empty() {
-            lines.push(metrics[0].get_header());
-            for metric in metrics {
-                lines.push(metric.to_string());
-            }
-        }
-    }
 }
 
 // Adds relevant metrics shown in a SHOW POOLS admin command.
 fn push_pool_stats(lines: &mut Vec<String>) {
-    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
     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_id.clone(), &name, value)
             {
-                grouped_metrics
+                lines.push(prometheus_metric.to_string());
-                    .entry(name)
-                    .or_default()
-                    .push(prometheus_metric);
             } else {
                 debug!("Metric {} not implemented for ({})", name, *pool_id);
             }
         }
     }
-    for (_key, metrics) in grouped_metrics {
-        if !metrics.is_empty() {
-            lines.push(metrics[0].get_header());
-            for metric in metrics {
-                lines.push(metric.to_string());
-            }
-        }
-    }
 }
 
 // Adds relevant metrics shown in a SHOW DATABASES admin command.
 fn push_database_stats(lines: &mut Vec<String>) {
-    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u32>>> = HashMap::new();
     for (_, pool) in get_all_pools() {
         let pool_config = pool.settings.clone();
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
                 let address = pool.address(shard, server);
                 let pool_state = pool.pool_state(shard, server);
+
                 let metrics = vec![
                     ("pool_size", pool_config.user.pool_size),
                     ("current_connections", pool_state.connections),
@@ -399,10 +316,7 @@ fn push_database_stats(lines: &mut Vec<String>) {
                     if let Some(prometheus_metric) =
                         PrometheusMetric::<u32>::from_database_info(address, key, value)
                     {
-                        grouped_metrics
+                        lines.push(prometheus_metric.to_string());
-                            .entry(key.to_string())
-                            .or_default()
-                            .push(prometheus_metric);
                     } else {
                         debug!("Metric {} not implemented for {}", key, address.name());
                     }
@@ -410,73 +324,45 @@ fn push_database_stats(lines: &mut Vec<String>) {
             }
         }
     }
-    for (_key, metrics) in grouped_metrics {
-        if !metrics.is_empty() {
-            lines.push(metrics[0].get_header());
-            for metric in metrics {
-                lines.push(metric.to_string());
-            }
-        }
-    }
 }
 
 // Adds relevant metrics shown in a SHOW SERVERS admin command.
 fn push_server_stats(lines: &mut Vec<String>) {
     let server_stats = get_server_stats();
-    let mut prom_stats = HashMap::<String, ServerPrometheusStats>::new();
+    let mut server_stats_by_addresses = HashMap::<String, Arc<ServerStats>>::new();
     for (_, stats) in server_stats {
-        let entry = prom_stats
+        server_stats_by_addresses.insert(stats.address_name(), stats);
-            .entry(stats.address_name())
-            .or_insert(ServerPrometheusStats {
-                bytes_received: 0,
-                bytes_sent: 0,
-                transaction_count: 0,
-                query_count: 0,
-                error_count: 0,
-                active_count: 0,
-                idle_count: 0,
-                login_count: 0,
-                tested_count: 0,
-            });
-        entry.bytes_received += stats.bytes_received.load(Ordering::Relaxed);
-        entry.bytes_sent += stats.bytes_sent.load(Ordering::Relaxed);
-        entry.transaction_count += stats.transaction_count.load(Ordering::Relaxed);
-        entry.query_count += stats.query_count.load(Ordering::Relaxed);
-        entry.error_count += stats.error_count.load(Ordering::Relaxed);
-        match stats.state.load(Ordering::Relaxed) {
-            crate::stats::ServerState::Login => entry.login_count += 1,
-            crate::stats::ServerState::Active => entry.active_count += 1,
-            crate::stats::ServerState::Tested => entry.tested_count += 1,
-            crate::stats::ServerState::Idle => entry.idle_count += 1,
-        }
     }
-    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
+
     for (_, pool) in get_all_pools() {
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
                 let address = pool.address(shard, server);
-                if let Some(server_info) = prom_stats.get(&address.name()) {
+                if let Some(server_info) = server_stats_by_addresses.get(&address.name()) {
                     let metrics = [
-                        ("bytes_received", server_info.bytes_received),
+                        (
-                        ("bytes_sent", server_info.bytes_sent),
+                            "bytes_received",
-                        ("transaction_count", server_info.transaction_count),
+                            server_info.bytes_received.load(Ordering::Relaxed),
-                        ("query_count", server_info.query_count),
+                        ),
-                        ("error_count", server_info.error_count),
+                        ("bytes_sent", server_info.bytes_sent.load(Ordering::Relaxed)),
-                        ("idle_count", server_info.idle_count),
+                        (
-                        ("active_count", server_info.active_count),
+                            "transaction_count",
-                        ("login_count", server_info.login_count),
+                            server_info.transaction_count.load(Ordering::Relaxed),
-                        ("tested_count", server_info.tested_count),
+                        ),
-                        ("is_banned", if pool.is_banned(address) { 1 } else { 0 }),
+                        (
-                        ("is_paused", if pool.paused() { 1 } else { 0 }),
+                            "query_count",
+                            server_info.query_count.load(Ordering::Relaxed),
+                        ),
+                        (
+                            "error_count",
+                            server_info.error_count.load(Ordering::Relaxed),
+                        ),
                     ];
                     for (key, value) in metrics {
                         if let Some(prometheus_metric) =
                             PrometheusMetric::<u64>::from_server_info(address, key, value)
                         {
-                            grouped_metrics
+                            lines.push(prometheus_metric.to_string());
-                                .entry(key.to_string())
-                                .or_default()
-                                .push(prometheus_metric);
                         } else {
                             debug!("Metric {} not implemented for {}", key, address.name());
                         }
@@ -485,46 +371,17 @@ fn push_server_stats(lines: &mut Vec<String>) {
             }
         }
     }
-    for (_key, metrics) in grouped_metrics {
-        if !metrics.is_empty() {
-            lines.push(metrics[0].get_header());
-            for metric in metrics {
-                lines.push(metric.to_string());
-            }
-        }
-    }
 }
 
 pub async fn start_metric_server(http_addr: SocketAddr) {
-    let listener = TcpListener::bind(http_addr);
+    let http_service_factory =
-    let listener = match listener.await {
+        make_service_fn(|_conn| async { Ok::<_, hyper::Error>(service_fn(prometheus_stats)) });
-        Ok(listener) => listener,
+    let server = Server::bind(&http_addr).serve(http_service_factory);
-        Err(e) => {
-            error!("Failed to bind prometheus server to HTTP address: {}.", e);
-            return;
-        }
-    };
     info!(
         "Exposing prometheus metrics on http://{}/metrics.",
         http_addr
     );
-    loop {
+    if let Err(e) = server.await {
-        let stream = match listener.accept().await {
+        error!("Failed to run HTTP server: {}.", e);
-            Ok((stream, _)) => stream,
-            Err(e) => {
-                error!("Error accepting connection: {}", e);
-                continue;
-            }
-        };
-        let io = TokioIo::new(stream);
-
-        tokio::task::spawn(async move {
-            if let Err(err) = http1::Builder::new()
-                .serve_connection(io, service_fn(prometheus_stats))
-                .await
-            {
-                eprintln!("Error serving HTTP connection for metrics: {:?}", err);
-            }
-        });
     }
 }

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("SCRAM".to_string()));
+            return Err(Error::ProtocolSyncError(format!("SCRAM")));
         }
 
         let salt = match general_purpose::STANDARD.decode(&server_message.salt) {
             Ok(salt) => salt,
-            Err(_) => return Err(Error::ProtocolSyncError("SCRAM".to_string())),
+            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
         };
 
         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("SCRAM".to_string())),
+            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
         };
 
         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("SCRAM".to_string()));
+            return Err(Error::ProtocolSyncError(format!("SCRAM")));
         }
 
         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("SCRAM".to_string())),
+            Err(_) => return Err(Error::ProtocolSyncError(format!("SCRAM"))),
         };
 
         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("SCRAM".to_string()));
+            return Err(Error::ProtocolSyncError(format!("SCRAM")));
         }
 
         Ok(FinalMessage {

src/server.rs

@@ -3,14 +3,11 @@
 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, HashSet, VecDeque};
+use std::collections::{BTreeSet, HashMap};
-use std::mem;
+use std::io::Read;
 use std::net::IpAddr;
-use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::SystemTime;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, BufStream};
@@ -18,11 +15,10 @@ use tokio::net::TcpStream;
 use tokio_rustls::rustls::{OwnedTrustAnchor, RootCertStore};
 use tokio_rustls::{client::TlsStream, TlsConnector};
 
-use crate::config::{get_config, Address, User};
+use crate::config::{get_config, get_prepared_statements_cache_size, 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;
@@ -109,10 +105,10 @@ impl StreamInner {
 
 #[derive(Copy, Clone)]
 struct CleanupState {
-    /// If server connection requires RESET ALL before checkin because of set statement
+    /// If server connection requires DISCARD ALL before checkin because of set statement
     needs_cleanup_set: bool,
 
-    /// If server connection requires DEALLOCATE ALL before checkin because of prepare statement
+    /// If server connection requires DISCARD ALL before checkin because of prepare statement
     needs_cleanup_prepare: bool,
 }
 
@@ -149,120 +145,6 @@ 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,
@@ -276,7 +158,7 @@ pub struct Server {
     buffer: BytesMut,
 
     /// Server information the server sent us over on startup.
-    server_parameters: ServerParameters,
+    server_info: BytesMut,
 
     /// Backend id and secret key used for query cancellation.
     process_id: i32,
@@ -294,7 +176,7 @@ pub struct Server {
     /// Is the server broken? We'll remote it from the pool if so.
     bad: bool,
 
-    /// If server connection requires reset statements before checkin
+    /// If server connection requires DISCARD ALL before checkin
     cleanup_state: CleanupState,
 
     /// Mapping of clients and servers used for query cancellation.
@@ -320,20 +202,13 @@ pub struct Server {
     /// 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_statements: BTreeSet<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,
@@ -342,8 +217,6 @@ impl Server {
         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;
@@ -443,7 +316,10 @@ impl Server {
 
                 // Something else?
                 m => {
-                    return Err(Error::SocketError(format!("Unknown message: {}", { m })));
+                    return Err(Error::SocketError(format!(
+                        "Unknown message: {}",
+                        m as char
+                    )));
                 }
             }
         } else {
@@ -461,22 +337,27 @@ impl Server {
             None => &user.username,
         };
 
-        let password = match user.server_password.as_ref() {
+        let password = match user.server_password {
-            Some(server_password) => Some(server_password),
+            Some(ref server_password) => Some(server_password),
-            None => user.password.as_ref(),
+            None => match user.password {
+                Some(ref password) => Some(password),
+                None => None,
+            },
         };
 
         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> = password.map(|password| ScramSha256::new(password));
+        let mut scram: Option<ScramSha256> = match password {
-
+            Some(password) => Some(ScramSha256::new(password)),
-        let mut server_parameters = ServerParameters::new();
+            None => None,
+        };
 
         loop {
             let code = match stream.read_u8().await {
@@ -707,7 +588,8 @@ impl Server {
 
                         // An error message will be present.
                         _ => {
-                            let mut error = vec![0u8; len as usize];
+                            // Read the error message without the terminating null character.
+                            let mut error = vec![0u8; len as usize - 4 - 1];
 
                             match stream.read_exact(&mut error).await {
                                 Ok(_) => (),
@@ -719,14 +601,10 @@ impl Server {
                                 }
                             };
 
-                            let fields = match PgErrorMsg::parse(&error) {
+                            // TODO: the error message contains multiple fields; we can decode them and
-                                Ok(f) => f,
+                            // present a prettier message to the user.
-                                Err(err) => {
+                            // See: https://www.postgresql.org/docs/12/protocol-error-fields.html
-                                    return Err(err);
+                            error!("Server error: {}", String::from_utf8_lossy(&error));
-                                }
-                            };
-                            trace!("error fields: {}", &fields);
-                            error!("server error: {}: {}", fields.severity, fields.message);
                         }
                     };
 
@@ -735,10 +613,9 @@ impl Server {
 
                 // ParameterStatus
                 'S' => {
-                    let mut bytes = BytesMut::with_capacity(len as usize - 4);
+                    let mut param = vec![0u8; len as usize - 4];
-                    bytes.resize(len as usize - mem::size_of::<i32>(), b'0');
 
-                    match stream.read_exact(&mut bytes[..]).await {
+                    match stream.read_exact(&mut param).await {
                         Ok(_) => (),
                         Err(_) => {
                             return Err(Error::ServerStartupError(
@@ -748,13 +625,12 @@ 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_parameters.set_param(key, value, true);
+                    server_info.put_u8(b'S');
+                    server_info.put_i32(len);
+                    server_info.put_slice(&param[..]);
                 }
 
                 // BackendKeyData
@@ -796,11 +672,11 @@ impl Server {
                         }
                     };
 
-                    let server = Server {
+                    let mut server = Server {
                         address: address.clone(),
                         stream: BufStream::new(stream),
                         buffer: BytesMut::with_capacity(8196),
-                        server_parameters,
+                        server_info,
                         process_id,
                         secret_key,
                         in_transaction: false,
@@ -812,7 +688,7 @@ impl Server {
                         addr_set,
                         connected_at: chrono::offset::Utc::now().naive_utc(),
                         stats,
-                        application_name: "pgcat".to_string(),
+                        application_name: String::new(),
                         last_activity: SystemTime::now(),
                         mirror_manager: match address.mirrors.len() {
                             0 => None,
@@ -823,16 +699,11 @@ impl Server {
                             )),
                         },
                         cleanup_connections,
-                        log_client_parameter_status_changes,
+                        prepared_statements: BTreeSet::new(),
-                        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);
                 }
 
@@ -882,7 +753,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();
@@ -902,10 +773,7 @@ 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(
+    pub async fn recv(&mut self) -> Result<BytesMut, Error> {
-        &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,
@@ -960,6 +828,7 @@ impl Server {
 
                     // There is no more data available from the server.
                     self.data_available = false;
+
                     break;
                 }
 
@@ -968,37 +837,6 @@ impl Server {
                     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
@@ -1007,24 +845,24 @@ impl Server {
                         self.in_copy_mode = false;
                     }
 
-                    match message.read_string() {
+                    let mut command_tag = String::new();
-                        Ok(command) => {
+                    match message.reader().read_to_string(&mut command_tag) {
+                        Ok(_) => {
                             // 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.as_str() {
+                            match command_tag.as_str() {
-                                "SET" => {
+                                "SET\0" => {
                                     // 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 reset statements sent
+                                    // This will reduce amount of discard 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;
                                 }
@@ -1038,20 +876,6 @@ 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.
@@ -1088,11 +912,6 @@ 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.
                 _ => (),
@@ -1114,103 +933,113 @@ impl Server {
         Ok(bytes)
     }
 
-    // Determines if the server already has a prepared statement with the given name
+    /// Add the prepared statement to being tracked by this server.
-    // Increments the prepared statement cache hit counter
+    /// The client is processing data that will create a prepared statement on this server.
-    pub fn has_prepared_statement(&mut self, name: &str) -> bool {
+    pub fn will_prepare(&mut self, name: &str) {
-        let cache = match &mut self.prepared_statement_cache {
+        debug!("Will prepare `{}`", name);
-            Some(cache) => cache,
-            None => return false,
-        };
 
-        let has_it = cache.get(name).is_some();
+        self.prepared_statements.insert(name.to_string());
-        if has_it {
+        self.stats.prepared_cache_add();
-            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> {
+    /// Check if we should prepare a statement on the server.
-        let cache = match &mut self.prepared_statement_cache {
+    pub fn should_prepare(&self, name: &str) -> bool {
-            Some(cache) => cache,
+        let should_prepare = !self.prepared_statements.contains(name);
-            None => return None,
-        };
 
+        debug!("Should prepare `{}`: {}", name, should_prepare);
+
+        if should_prepare {
+            self.stats.prepared_cache_miss();
+        } else {
+            self.stats.prepared_cache_hit();
+        }
+
+        should_prepare
+    }
+
+    /// Create a prepared statement on the server.
+    pub async fn prepare(&mut self, parse: &Parse) -> Result<(), Error> {
+        debug!("Preparing `{}`", parse.name);
+
+        let bytes: BytesMut = parse.try_into()?;
+        self.send(&bytes).await?;
+        self.send(&flush()).await?;
+
+        // Read and discard ParseComplete (B)
+        match read_message(&mut self.stream).await {
+            Ok(_) => (),
+            Err(err) => {
+                self.bad = true;
+                return Err(err);
+            }
+        }
+
+        self.prepared_statements.insert(parse.name.to_string());
         self.stats.prepared_cache_add();
 
-        // If we evict something, we need to close it on the server
+        debug!("Prepared `{}`", parse.name);
-        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
+        Ok(())
     }
 
-    fn remove_prepared_statement_from_cache(&mut self, name: &str) {
+    /// Maintain adequate cache size on the server.
-        let cache = match &mut self.prepared_statement_cache {
+    pub async fn maintain_cache(&mut self) -> Result<(), Error> {
-            Some(cache) => cache,
+        debug!("Cache maintenance run");
-            None => return,
-        };
 
-        self.stats.prepared_cache_remove();
+        let max_cache_size = get_prepared_statements_cache_size();
-        cache.pop(name);
+        let mut names = Vec::new();
-    }
 
-    pub async fn register_prepared_statement(
+        while self.prepared_statements.len() >= max_cache_size {
-        &mut self,
+            // The prepared statmeents are alphanumerically sorted by the BTree.
-        parse: &Parse,
+            // FIFO.
-        should_send_parse_to_server: bool,
+            if let Some(name) = self.prepared_statements.pop_last() {
-    ) -> Result<(), Error> {
+                names.push(name);
-        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(())
         }
+
+        self.deallocate(names).await?;
+
+        Ok(())
+    }
+
+    /// Remove the prepared statement from being tracked by this server.
+    /// The client is processing data that will cause the server to close the prepared statement.
+    pub fn will_close(&mut self, name: &str) {
+        debug!("Will close `{}`", name);
+
+        self.prepared_statements.remove(name);
+    }
+
+    /// Close a prepared statement on the server.
+    pub async fn deallocate(&mut self, names: Vec<String>) -> Result<(), Error> {
+        for name in &names {
+            debug!("Deallocating prepared statement `{}`", name);
+
+            let close = Close::new(name);
+            let bytes: BytesMut = close.try_into()?;
+
+            self.send(&bytes).await?;
+        }
+
+        self.send(&flush()).await?;
+
+        // Read and discard CloseComplete (3)
+        for name in &names {
+            match read_message(&mut self.stream).await {
+                Ok(_) => {
+                    self.prepared_statements.remove(name);
+                    self.stats.prepared_cache_remove();
+                    debug!("Closed `{}`", name);
+                }
+
+                Err(err) => {
+                    self.bad = true;
+                    return Err(err);
+                }
+            };
+        }
+
+        Ok(())
     }
 
     /// If the server is still inside a transaction.
@@ -1220,7 +1049,6 @@ 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
     }
@@ -1254,33 +1082,14 @@ impl Server {
     }
 
     /// Get server startup information to forward it to the client.
-    pub fn server_parameters(&self) -> ServerParameters {
+    /// Not used at the moment.
-        self.server_parameters.clone()
+    pub fn server_info(&self) -> BytesMut {
-    }
+        self.server_info.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.
-    pub fn mark_bad(&mut self, reason: &str) {
+    pub fn mark_bad(&mut self) {
-        error!("Server {:?} marked bad, reason: {}", self.address, reason);
+        error!("Server {:?} marked bad", self.address);
         self.bad = true;
     }
 
@@ -1309,7 +1118,7 @@ impl Server {
         self.send(&query).await?;
 
         loop {
-            let _ = self.recv(None).await?;
+            let _ = self.recv().await?;
 
             if !self.data_available {
                 break;
@@ -1327,42 +1136,47 @@ 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!(target: "pgcat::server::cleanup", "Server returned while still in transaction, rolling back transaction");
+            warn!("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 `RESET ALL` if we think the session is altered instead of just sending
+        // send `DISCARD ALL` if we think the session is altered instead of just sending
         // it before each checkin.
         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);
+            warn!("Server returned with session state altered, discarding state ({}) for application {}", self.cleanup_state, self.application_name);
-            let mut reset_string = String::from("RESET ROLE;");
+            self.query("DISCARD ALL").await?;
-
+            self.query("RESET ROLE").await?;
-            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();
         }
 
         if self.in_copy_mode() {
-            warn!(target: "pgcat::server::cleanup", "Server returned while still in copy-mode");
+            warn!("Server returned while still in copy-mode");
         }
 
         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(())
+        }
+    }
+
     /// get Server stats
     pub fn stats(&self) -> Arc<ServerStats> {
         self.stats.clone()
@@ -1379,20 +1193,22 @@ impl Server {
         self.last_activity
     }
 
-    // Marks a connection as needing cleanup at checkin
+    // Marks a connection as needing DISCARD ALL at checkin
     pub fn mark_dirty(&mut self) {
         self.cleanup_state.set_true();
     }
 
     pub fn mirror_send(&mut self, bytes: &BytesMut) {
-        if let Some(manager) = self.mirror_manager.as_mut() {
+        match self.mirror_manager.as_mut() {
-            manager.send(bytes)
+            Some(manager) => manager.send(bytes),
+            None => (),
         }
     }
 
     pub fn mirror_disconnect(&mut self) {
-        if let Some(manager) = self.mirror_manager.as_mut() {
+        match self.mirror_manager.as_mut() {
-            manager.disconnect()
+            Some(manager) => manager.disconnect(),
+            None => (),
         }
     }
 
@@ -1414,15 +1230,13 @@ impl Server {
             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(None).await?;
+        let mut message = server.recv().await?;
 
-        parse_query_message(&mut message).await
+        Ok(parse_query_message(&mut message).await?)
     }
 }
 

src/sharding.rs

@@ -14,11 +14,11 @@ pub enum ShardingFunction {
     Sha1,
 }
 
-impl std::fmt::Display for ShardingFunction {
+impl ToString for ShardingFunction {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    fn to_string(&self) -> String {
-        match self {
+        match *self {
-            ShardingFunction::PgBigintHash => write!(f, "pg_bigint_hash"),
+            ShardingFunction::PgBigintHash => "pg_bigint_hash".to_string(),
-            ShardingFunction::Sha1 => write!(f, "sha1"),
+            ShardingFunction::Sha1 => "sha1".to_string(),
         }
     }
 }
@@ -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 = [
+        let ids = vec![
             0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
         ];
-        let shards = [
+        let shards = vec![
             4, 7, 8, 3, 6, 0, 0, 10, 3, 11, 1, 7, 4, 4, 11, 2, 5, 0, 8, 3,
         ];
 

src/stats/client.rs

@@ -41,11 +41,6 @@ pub struct ClientStats {
     /// Maximum time spent waiting for a connection from pool, measures in microseconds
     pub max_wait_time: Arc<AtomicU64>,
 
-    // Time when the client started waiting for a connection from pool, measures in microseconds
-    // We use connect_time as the reference point for this value
-    // U64 can represent ~5850 centuries in microseconds, so we should be fine
-    pub wait_start_us: Arc<AtomicU64>,
-
     /// Current state of the client
     pub state: Arc<AtomicClientState>,
 
@@ -69,7 +64,6 @@ impl Default for ClientStats {
             pool_name: String::new(),
             total_wait_time: Arc::new(AtomicU64::new(0)),
             max_wait_time: Arc::new(AtomicU64::new(0)),
-            wait_start_us: 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)),
@@ -117,9 +111,6 @@ impl ClientStats {
 
     /// Reports a client is waiting for a connection
     pub fn waiting(&self) {
-        let wait_start = self.connect_time.elapsed().as_micros() as u64;
-
-        self.wait_start_us.store(wait_start, Ordering::Relaxed);
         self.state.store(ClientState::Waiting, Ordering::Relaxed);
     }
 
@@ -131,13 +122,6 @@ impl ClientStats {
     /// Reports a client has failed to obtain a connection from a connection pool
     pub fn checkout_error(&self) {
         self.state.store(ClientState::Idle, Ordering::Relaxed);
-        self.update_wait_times();
-    }
-
-    /// Reports a client has succeeded in obtaining a connection from a connection pool
-    pub fn checkout_success(&self) {
-        self.state.store(ClientState::Active, Ordering::Relaxed);
-        self.update_wait_times();
     }
 
     /// Reports a client has had the server assigned to it be banned
@@ -146,26 +130,12 @@ impl ClientStats {
         self.error_count.fetch_add(1, Ordering::Relaxed);
     }
 
-    fn update_wait_times(&self) {
+    /// Reporters the time spent by a client waiting to get a healthy connection from the pool
-        if self.wait_start_us.load(Ordering::Relaxed) == 0 {
+    pub fn checkout_time(&self, microseconds: u64) {
-            return;
-        }
-
-        let wait_time_us = self.get_current_wait_time_us();
         self.total_wait_time
-            .fetch_add(wait_time_us, Ordering::Relaxed);
+            .fetch_add(microseconds, Ordering::Relaxed);
         self.max_wait_time
-            .fetch_max(wait_time_us, Ordering::Relaxed);
+            .fetch_max(microseconds, Ordering::Relaxed);
-        self.wait_start_us.store(0, Ordering::Relaxed);
-    }
-
-    pub fn get_current_wait_time_us(&self) -> u64 {
-        let wait_start_us = self.wait_start_us.load(Ordering::Relaxed);
-        let microseconds_since_connection_epoch = self.connect_time.elapsed().as_micros() as u64;
-        if wait_start_us == 0 || microseconds_since_connection_epoch < wait_start_us {
-            return 0;
-        }
-        microseconds_since_connection_epoch - wait_start_us
     }
 
     /// Report a query executed by a client against a server

src/stats/pool.rs

@@ -64,11 +64,8 @@ impl PoolStats {
                         ClientState::Idle => pool_stats.cl_idle += 1,
                         ClientState::Waiting => pool_stats.cl_waiting += 1,
                     }
-                    let wait_start_us = client.wait_start_us.load(Ordering::Relaxed);
+                    let max_wait = client.max_wait_time.load(Ordering::Relaxed);
-                    if wait_start_us > 0 {
+                    pool_stats.maxwait = std::cmp::max(pool_stats.maxwait, max_wait);
-                        let wait_time_us = client.get_current_wait_time_us();
-                        pool_stats.maxwait = std::cmp::max(pool_stats.maxwait, wait_time_us);
-                    }
                 }
                 None => debug!("Client from an obselete pool"),
             }
@@ -89,11 +86,11 @@ impl PoolStats {
             }
         }
 
-        map
+        return map;
     }
 
     pub fn generate_header() -> Vec<(&'static str, DataType)> {
-        vec![
+        return vec![
             ("database", DataType::Text),
             ("user", DataType::Text),
             ("pool_mode", DataType::Text),
@@ -108,11 +105,11 @@ impl PoolStats {
             ("sv_login", DataType::Numeric),
             ("maxwait", DataType::Numeric),
             ("maxwait_us", DataType::Numeric),
-        ]
+        ];
     }
 
     pub fn generate_row(&self) -> Vec<String> {
-        vec![
+        return vec![
             self.identifier.db.clone(),
             self.identifier.user.clone(),
             self.mode.to_string(),
@@ -127,7 +124,7 @@ impl PoolStats {
             self.sv_login.to_string(),
             (self.maxwait / 1_000_000).to_string(),
             (self.maxwait % 1_000_000).to_string(),
-        ]
+        ];
     }
 }
 

src/stats/server.rs

@@ -49,7 +49,6 @@ pub struct ServerStats {
     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>,
 }
 
@@ -69,7 +68,6 @@ impl Default for ServerStats {
             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)),
         }
     }
@@ -223,7 +221,6 @@ impl ServerStats {
     }
 
     pub fn prepared_cache_remove(&self) {
-        self.prepared_eviction_count.fetch_add(1, Ordering::Relaxed);
         self.prepared_cache_size.fetch_sub(1, Ordering::Relaxed);
     }
 }

start_test_env.sh

@@ -1,34 +0,0 @@
-GREEN="\033[0;32m"
-RED="\033[0;31m"
-BLUE="\033[0;34m"
-RESET="\033[0m"
-
-
-cd tests/docker/
-docker compose kill main || true
-docker compose build main
-docker compose down
-docker compose up -d
-# wait for the container to start
-while ! docker compose exec main ls; do
-    echo "Waiting for test environment to start"
-    sleep 1
-done
-echo "==================================="
-docker compose exec -e LOG_LEVEL=error -d main toxiproxy-server
-docker compose exec --workdir /app main cargo build
-docker compose exec -d --workdir /app main ./target/debug/pgcat ./.circleci/pgcat.toml
-docker compose exec --workdir /app/tests/ruby main bundle install
-docker compose exec --workdir /app/tests/python main pip3 install -r requirements.txt
-echo "Interactive test environment ready"
-echo "To run integration tests, you can use the following commands:"
-echo -e "   ${BLUE}Ruby:   ${RED}cd /app/tests/ruby && bundle exec ruby tests.rb --format documentation${RESET}"
-echo -e "   ${BLUE}Python: ${RED}cd /app/ && pytest ${RESET}"
-echo -e "   ${BLUE}Rust:   ${RED}cd /app/tests/rust && cargo run ${RESET}"
-echo -e "   ${BLUE}Go:     ${RED}cd /app/tests/go && /usr/local/go/bin/go test${RESET}"
-echo "the source code for tests are directly linked to the source code in the container so you can modify the code and run the tests again"
-echo "You can rebuild PgCat from within the container by running"
-echo -e "  ${GREEN}cargo build${RESET}"
-echo "and then run the tests again"
-echo "==================================="
-docker compose exec --workdir /app/tests main bash

tests/docker/Dockerfile

@@ -1,13 +1,8 @@
 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/docker/docker-compose.yml

@@ -1,3 +1,4 @@
+version: "3"
 services:
   pg1:
     image: postgres:14
@@ -47,8 +48,6 @@ services:
   main:
     build: .
     command: ["bash", "/app/tests/docker/run.sh"]
-    environment:
-      - INTERACTIVE_TEST_ENVIRONMENT=true
     volumes:
       - ../../:/app/
       - /app/target/

tests/docker/run.sh

@@ -5,38 +5,6 @@ rm /app/*.profraw || true
 rm /app/pgcat.profdata || true
 rm -rf /app/cov || true
 
-# Prepares the interactive test environment
-# 
-if [ -n "$INTERACTIVE_TEST_ENVIRONMENT" ]; then
-    ports=(5432 7432 8432 9432 10432)
-    for port in "${ports[@]}"; do
-        is_it_up=0
-        attempts=0
-        while [ $is_it_up -eq 0 ]; do
-            PGPASSWORD=postgres psql -h 127.0.0.1 -p $port -U postgres -c '\q' > /dev/null 2>&1
-            if [ $? -eq 0 ]; then
-                echo "PostgreSQL on port $port is up."
-                is_it_up=1
-            else
-                attempts=$((attempts+1))
-                if [ $attempts -gt 10 ]; then
-                    echo "PostgreSQL on port $port is down, giving up."
-                    exit 1
-                fi
-                echo "PostgreSQL on port $port is down, waiting for it to start."
-                sleep 1
-            fi
-        done
-    done
-    PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 5432  -U postgres -f /app/tests/sharding/query_routing_setup.sql
-    PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 7432  -U postgres -f /app/tests/sharding/query_routing_setup.sql
-    PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 8432  -U postgres -f /app/tests/sharding/query_routing_setup.sql
-    PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 9432  -U postgres -f /app/tests/sharding/query_routing_setup.sql
-    PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 10432 -U postgres -f /app/tests/sharding/query_routing_setup.sql
-    sleep 100000000000000000
-    exit 0
-fi
-
 export LLVM_PROFILE_FILE="/app/pgcat-%m-%p.profraw"
 export RUSTC_BOOTSTRAP=1
 export CARGO_INCREMENTAL=0

tests/go/go.mod

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

tests/go/go.sum

@@ -1,2 +0,0 @@
-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

@@ -1,162 +0,0 @@
-#
-# 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

@@ -1,52 +0,0 @@
-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

@@ -1,81 +0,0 @@
-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/python/requirements.txt

@@ -1,3 +1,2 @@
-pytest
 psycopg2==2.9.3
 psutil==5.9.1

tests/python/test_auth.py

@@ -1,71 +0,0 @@
-import utils
-import signal
-
-class TestTrustAuth:
-    @classmethod
-    def setup_method(cls):
-        config= """
-        [general]
-        host = "0.0.0.0"
-        port = 6432
-        admin_username = "admin_user"
-        admin_password = ""
-        admin_auth_type = "trust"
-
-        [pools.sharded_db.users.0]
-        username = "sharding_user"
-        password = "sharding_user"
-        auth_type = "trust"
-        pool_size = 10
-        min_pool_size = 1
-        pool_mode = "transaction"
-
-        [pools.sharded_db.shards.0]
-        servers = [
-          [ "127.0.0.1", 5432, "primary" ],
-        ]
-        database = "shard0"
-        """
-        utils.pgcat_generic_start(config)
-
-    @classmethod
-    def teardown_method(self):
-        utils.pg_cat_send_signal(signal.SIGTERM)
-
-    def test_admin_trust_auth(self):
-        conn, cur = utils.connect_db_trust(admin=True)
-        cur.execute("SHOW POOLS")
-        res = cur.fetchall()
-        print(res)
-        utils.cleanup_conn(conn, cur)
-
-    def test_normal_trust_auth(self):
-        conn, cur = utils.connect_db_trust(autocommit=False)
-        cur.execute("SELECT 1")
-        res = cur.fetchall()
-        print(res)
-        utils.cleanup_conn(conn, cur)
-
-class TestMD5Auth:
-    @classmethod
-    def setup_method(cls):
-        utils.pgcat_start()
-
-    @classmethod
-    def teardown_method(self):
-        utils.pg_cat_send_signal(signal.SIGTERM)
-
-    def test_normal_db_access(self):
-        conn, cur = utils.connect_db(autocommit=False)
-        cur.execute("SELECT 1")
-        res = cur.fetchall()
-        print(res)
-        utils.cleanup_conn(conn, cur)
-
-    def test_admin_db_access(self):
-        conn, cur = utils.connect_db(admin=True)
-
-        cur.execute("SHOW POOLS")
-        res = cur.fetchall()
-        print(res)
-        utils.cleanup_conn(conn, cur)

tests/python/tests.py

@@ -1,12 +1,84 @@
-
+from typing import Tuple
+import psycopg2
+import psutil
+import os
 import signal
 import time
 
-import psycopg2
-import utils
-
 SHUTDOWN_TIMEOUT = 5
 
+PGCAT_HOST = "127.0.0.1"
+PGCAT_PORT = "6432"
+
+
+def pgcat_start():
+    pg_cat_send_signal(signal.SIGTERM)
+    os.system("./target/debug/pgcat .circleci/pgcat.toml &")
+    time.sleep(2)
+
+
+def pg_cat_send_signal(signal: signal.Signals):
+    try:
+        for proc in psutil.process_iter(["pid", "name"]):
+            if "pgcat" == proc.name():
+                os.kill(proc.pid, signal)
+    except Exception as e:
+        # The process can be gone when we send this signal
+        print(e)
+
+    if signal == signal.SIGTERM:
+        # Returns 0 if pgcat process exists
+        time.sleep(2)
+        if not os.system('pgrep pgcat'):
+            raise Exception("pgcat not closed after SIGTERM")
+
+
+def connect_db(
+    autocommit: bool = True,
+    admin: bool = False,
+) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
+
+    if admin:
+        user = "admin_user"
+        password = "admin_pass"
+        db = "pgcat"
+    else:
+        user = "sharding_user"
+        password = "sharding_user"
+        db = "sharded_db"
+
+    conn = psycopg2.connect(
+        f"postgres://{user}:{password}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
+        connect_timeout=2,
+    )
+    conn.autocommit = autocommit
+    cur = conn.cursor()
+
+    return (conn, cur)
+
+
+def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
+    cur.close()
+    conn.close()
+
+
+def test_normal_db_access():
+    pgcat_start()
+    conn, cur = connect_db(autocommit=False)
+    cur.execute("SELECT 1")
+    res = cur.fetchall()
+    print(res)
+    cleanup_conn(conn, cur)
+
+
+def test_admin_db_access():
+    conn, cur = connect_db(admin=True)
+
+    cur.execute("SHOW POOLS")
+    res = cur.fetchall()
+    print(res)
+    cleanup_conn(conn, cur)
+
 
 def test_shutdown_logic():
 
@@ -14,17 +86,17 @@ def test_shutdown_logic():
     # NO ACTIVE QUERIES SIGINT HANDLING
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and send query (not in transaction)
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
     cur.execute("COMMIT;")
 
     # Send sigint to pgcat
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     # Check that any new queries fail after sigint since server should close with no active transactions
@@ -36,18 +108,18 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint")
 
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # NO ACTIVE QUERIES ADMIN SHUTDOWN COMMAND
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    admin_conn, admin_cur = connect_db(admin=True)
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
@@ -66,24 +138,24 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint")
 
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
+    cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE TRANSACTION WITH SIGINT
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     # Check that any new queries succeed after sigint since server should still allow transaction to complete
@@ -93,18 +165,18 @@ def test_shutdown_logic():
         # Fail if query fails since server closed
         raise Exception("Server closed while in transaction", e.pgerror)
 
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE TRANSACTION WITH ADMIN SHUTDOWN COMMAND
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    admin_conn, admin_cur = connect_db(admin=True)
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
@@ -122,30 +194,30 @@ def test_shutdown_logic():
         # Fail if query fails since server closed
         raise Exception("Server closed while in transaction", e.pgerror)
 
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
+    cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # NO NEW NON-ADMIN CONNECTIONS DURING SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
 
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     start = time.perf_counter()
     try:
-        conn, cur = utils.connect_db()
+        conn, cur = connect_db()
         cur.execute("SELECT 1;")
-        utils.cleanup_conn(conn, cur)
+        cleanup_conn(conn, cur)
     except psycopg2.OperationalError as e:
         time_taken = time.perf_counter() - start
         if time_taken > 0.1:
@@ -155,49 +227,49 @@ def test_shutdown_logic():
     else:
         raise Exception("Able connect to database during shutdown")
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
+    cleanup_conn(transaction_conn, transaction_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # ALLOW NEW ADMIN CONNECTIONS DURING SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
 
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     try:
-        conn, cur = utils.connect_db(admin=True)
+        conn, cur = connect_db(admin=True)
         cur.execute("SHOW DATABASES;")
-        utils.cleanup_conn(conn, cur)
+        cleanup_conn(conn, cur)
     except psycopg2.OperationalError as e:
         raise Exception(e)
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
+    cleanup_conn(transaction_conn, transaction_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # ADMIN CONNECTIONS CONTINUING TO WORK AFTER SHUTDOWN
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction
-    transaction_conn, transaction_cur = utils.connect_db()
+    transaction_conn, transaction_cur = connect_db()
     transaction_cur.execute("BEGIN;")
     transaction_cur.execute("SELECT 1;")
 
-    admin_conn, admin_cur = utils.connect_db(admin=True)
+    admin_conn, admin_cur = connect_db(admin=True)
     admin_cur.execute("SHOW DATABASES;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
     time.sleep(1)
 
     try:
@@ -205,24 +277,24 @@ def test_shutdown_logic():
     except psycopg2.OperationalError as e:
         raise Exception("Could not execute admin command:", e)
 
-    utils.cleanup_conn(transaction_conn, transaction_cur)
+    cleanup_conn(transaction_conn, transaction_cur)
-    utils.cleanup_conn(admin_conn, admin_cur)
+    cleanup_conn(admin_conn, admin_cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
     # HANDLE SHUTDOWN TIMEOUT WITH SIGINT
 
     # Start pgcat
-    utils.pgcat_start()
+    pgcat_start()
 
     # Create client connection and begin transaction, which should prevent server shutdown unless shutdown timeout is reached
-    conn, cur = utils.connect_db()
+    conn, cur = connect_db()
 
     cur.execute("BEGIN;")
     cur.execute("SELECT 1;")
 
     # Send sigint to pgcat while still in transaction
-    utils.pg_cat_send_signal(signal.SIGINT)
+    pg_cat_send_signal(signal.SIGINT)
 
     # pgcat shutdown timeout is set to SHUTDOWN_TIMEOUT seconds, so we sleep for SHUTDOWN_TIMEOUT + 1 seconds
     time.sleep(SHUTDOWN_TIMEOUT + 1)
@@ -236,7 +308,12 @@ def test_shutdown_logic():
         # Fail if query execution succeeded
         raise Exception("Server not closed after sigint and expected timeout")
 
-    utils.cleanup_conn(conn, cur)
+    cleanup_conn(conn, cur)
-    utils.pg_cat_send_signal(signal.SIGTERM)
+    pg_cat_send_signal(signal.SIGTERM)
 
     # - - - - - - - - - - - - - - - - - -
+
+
+test_normal_db_access()
+test_admin_db_access()
+test_shutdown_logic()

tests/python/utils.py

@@ -1,110 +0,0 @@
-import os
-import signal
-import time
-from typing import Tuple
-import tempfile
-
-import psutil
-import psycopg2
-
-PGCAT_HOST = "127.0.0.1"
-PGCAT_PORT = "6432"
-
-
-def _pgcat_start(config_path: str):
-    pg_cat_send_signal(signal.SIGTERM)
-    os.system(f"./target/debug/pgcat {config_path} &")
-    time.sleep(2)
-
-
-def pgcat_start():
-    _pgcat_start(config_path='.circleci/pgcat.toml')
-
-
-def pgcat_generic_start(config: str):
-    tmp = tempfile.NamedTemporaryFile()
-    with open(tmp.name, 'w') as f:
-        f.write(config)
-    _pgcat_start(config_path=tmp.name)
-
-
-def glauth_send_signal(signal: signal.Signals):
-    try:
-        for proc in psutil.process_iter(["pid", "name"]):
-            if proc.name() == "glauth":
-                os.kill(proc.pid, signal)
-    except Exception as e:
-        # The process can be gone when we send this signal
-        print(e)
-
-    if signal == signal.SIGTERM:
-        # Returns 0 if pgcat process exists
-        time.sleep(2)
-        if not os.system('pgrep glauth'):
-            raise Exception("glauth not closed after SIGTERM")
-
-
-def pg_cat_send_signal(signal: signal.Signals):
-    try:
-        for proc in psutil.process_iter(["pid", "name"]):
-            if "pgcat" == proc.name():
-                os.kill(proc.pid, signal)
-    except Exception as e:
-        # The process can be gone when we send this signal
-        print(e)
-
-    if signal == signal.SIGTERM:
-        # Returns 0 if pgcat process exists
-        time.sleep(2)
-        if not os.system('pgrep pgcat'):
-            raise Exception("pgcat not closed after SIGTERM")
-
-
-def connect_db(
-    autocommit: bool = True,
-    admin: bool = False,
-) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
-
-    if admin:
-        user = "admin_user"
-        password = "admin_pass"
-        db = "pgcat"
-    else:
-        user = "sharding_user"
-        password = "sharding_user"
-        db = "sharded_db"
-
-    conn = psycopg2.connect(
-        f"postgres://{user}:{password}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
-        connect_timeout=2,
-    )
-    conn.autocommit = autocommit
-    cur = conn.cursor()
-
-    return (conn, cur)
-
-def connect_db_trust(
-    autocommit: bool = True,
-    admin: bool = False,
-) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
-
-    if admin:
-        user = "admin_user"
-        db = "pgcat"
-    else:
-        user = "sharding_user"
-        db = "sharded_db"
-
-    conn = psycopg2.connect(
-        f"postgres://{user}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
-        connect_timeout=2,
-    )
-    conn.autocommit = autocommit
-    cur = conn.cursor()
-
-    return (conn, cur)
-
-
-def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
-    cur.close()
-    conn.close()

tests/ruby/Gemfile.lock

@@ -1,33 +1,22 @@
 GEM
   remote: https://rubygems.org/
   specs:
-    activemodel (7.1.4)
+    activemodel (7.0.4.1)
-      activesupport (= 7.1.4)
+      activesupport (= 7.0.4.1)
-    activerecord (7.1.4)
+    activerecord (7.0.4.1)
-      activemodel (= 7.1.4)
+      activemodel (= 7.0.4.1)
-      activesupport (= 7.1.4)
+      activesupport (= 7.0.4.1)
-      timeout (>= 0.4.0)
+    activesupport (7.0.4.1)
-    activesupport (7.1.4)
-      base64
-      bigdecimal
       concurrent-ruby (~> 1.0, >= 1.0.2)
-      connection_pool (>= 2.2.5)
-      drb
       i18n (>= 1.6, < 2)
       minitest (>= 5.1)
-      mutex_m
       tzinfo (~> 2.0)
     ast (2.4.2)
-    base64 (0.2.0)
+    concurrent-ruby (1.1.10)
-    bigdecimal (3.1.8)
-    concurrent-ruby (1.3.4)
-    connection_pool (2.4.1)
     diff-lcs (1.5.0)
-    drb (2.2.1)
+    i18n (1.12.0)
-    i18n (1.14.5)
       concurrent-ruby (~> 1.0)
-    minitest (5.25.1)
+    minitest (5.17.0)
-    mutex_m (0.2.0)
     parallel (1.22.1)
     parser (3.1.2.0)
       ast (~> 2.4.1)
@@ -35,8 +24,7 @@ GEM
     pg (1.3.2)
     rainbow (3.1.1)
     regexp_parser (2.3.1)
-    rexml (3.3.6)
+    rexml (3.2.5)
-      strscan
     rspec (3.11.0)
       rspec-core (~> 3.11.0)
       rspec-expectations (~> 3.11.0)
@@ -62,12 +50,10 @@ GEM
     rubocop-ast (1.17.0)
       parser (>= 3.1.1.0)
     ruby-progressbar (1.11.0)
-    strscan (3.1.0)
-    timeout (0.4.1)
     toml (0.3.0)
       parslet (>= 1.8.0, < 3.0.0)
     toxiproxy (2.0.1)
-    tzinfo (2.0.6)
+    tzinfo (2.0.5)
       concurrent-ruby (~> 1.0)
     unicode-display_width (2.1.0)
 

tests/ruby/admin_spec.rb

@@ -90,49 +90,4 @@ describe "Admin" do
       expect(results["pool_mode"]).to eq("transaction")
     end
   end
-
-  [
-    "SHOW ME THE MONEY", 
-    "SHOW ME THE WAY", 
-    "SHOW UP", 
-    "SHOWTIME",
-    "HAMMER TIME",
-    "SHOWN TO BE TRUE",
-    "SHOW ",
-    "SHOW     ",
-    "SHOW 1",
-    ";;;;;"
-  ].each do |cmd|
-    describe "Bad command #{cmd}" do
-      it "does not panic and responds with PG::SystemError" do
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-        expect { admin_conn.async_exec(cmd) }.to raise_error(PG::SystemError).with_message(/Unsupported/)
-        admin_conn.close
-      end
-    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/helpers/auth_query_helper.rb

@@ -33,15 +33,15 @@ module Helpers
             "0" => {
               "database" => "shard0",
               "servers" => [
-                ["localhost", primary.port.to_i, "primary"],
+                ["localhost", primary.port.to_s, "primary"],
-                ["localhost", replica.port.to_i, "replica"],
+                ["localhost", replica.port.to_s, "replica"],
               ]
             },
           },
           "users" => { "0" => user.merge(config_user) }
         }
       }
-      pgcat_cfg["general"]["port"] = pgcat.port.to_i
+      pgcat_cfg["general"]["port"] = pgcat.port
       pgcat.update_config(pgcat_cfg)
       pgcat.start
       
@@ -92,8 +92,8 @@ module Helpers
             "0" => {
               "database" => database,
               "servers" => [
-                ["localhost", primary.port.to_i, "primary"],
+                ["localhost", primary.port.to_s, "primary"],
-                ["localhost", replica.port.to_i, "replica"],
+                ["localhost", replica.port.to_s, "replica"],
               ]
             },
           },

tests/ruby/helpers/pg_instance.rb

@@ -7,24 +7,10 @@ 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.to_i
+    @original_port = port
     @toxiproxy_port = 10000 + port.to_i
-    @port = @toxiproxy_port.to_i
+    @port = @toxiproxy_port
 
     @username = username
     @password = password
@@ -62,9 +48,9 @@ class PgInstance
 
   def take_down
     if block_given?
-      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 1).apply { yield }
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).apply { yield }
     else
-      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 1).toxics.each(&:save)
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).toxics.each(&:save)
     end
   end
 
@@ -103,6 +89,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 LIKE '%SELECT $1 + $2%'")[0]["sum"].to_i }
+    with_connection { |c| c.async_exec("SELECT SUM(calls) FROM pg_stat_statements WHERE query = 'SELECT $1 + $2'")[0]["sum"].to_i }
   end
 end

tests/ruby/helpers/pgcat_helper.rb

@@ -34,13 +34,12 @@ 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_i, "primary"]] },
+            "0" => { "database" => "shard0", "servers" => [["localhost", primary0.port.to_s, "primary"]] },
-            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_i, "primary"]] },
+            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_s, "primary"]] },
-            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_i, "primary"]] },
+            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_s, "primary"]] },
           },
           "users" => { "0" => user },
           "plugins" => {
@@ -100,7 +99,7 @@ module Helpers
             "0" => {
               "database" => "shard0",
               "servers" => [
-                ["localhost", primary.port.to_i, "primary"]
+                ["localhost", primary.port.to_s, "primary"]
               ]
             },
           },
@@ -146,10 +145,10 @@ module Helpers
           "0" => {
             "database" => "shard0",
             "servers" => [
-              ["localhost", primary.port.to_i, "primary"],
+              ["localhost", primary.port.to_s, "primary"],
-              ["localhost", replica0.port.to_i, "replica"],
+              ["localhost", replica0.port.to_s, "replica"],
-              ["localhost", replica1.port.to_i, "replica"],
+              ["localhost", replica1.port.to_s, "replica"],
-              ["localhost", replica2.port.to_i, "replica"]
+              ["localhost", replica2.port.to_s, "replica"]
             ]
           },
         },

tests/ruby/helpers/pgcat_process.rb

@@ -1,10 +1,8 @@
 require 'pg'
-require 'json'
+require 'toml'
-require 'tempfile'
 require 'fileutils'
 require 'securerandom'
 
-class ConfigReloadFailed < StandardError; end
 class PgcatProcess
   attr_reader :port
   attr_reader :pid
@@ -20,7 +18,7 @@ class PgcatProcess
   end
 
   def initialize(log_level)
-    @env = {}
+    @env = {"RUST_LOG" => log_level}
     @port = rand(20000..32760)
     @log_level = log_level
     @log_filename = "/tmp/pgcat_log_#{SecureRandom.urlsafe_base64}.log"
@@ -32,7 +30,7 @@ class PgcatProcess
                      '../../target/debug/pgcat'
                    end
 
-    @command = "#{command_path} #{@config_filename} --log-level #{@log_level}"
+    @command = "#{command_path} #{@config_filename}"
 
     FileUtils.cp("../../pgcat.toml", @config_filename)
     cfg = current_config
@@ -48,34 +46,22 @@ class PgcatProcess
 
   def update_config(config_hash)
     @original_config = current_config
-    Tempfile.create('json_out', '/tmp') do |f|
+    output_to_write = TOML::Generator.new(config_hash).body
-      f.write(config_hash.to_json)
+    output_to_write = output_to_write.gsub(/,\s*["|'](\d+)["|']\s*,/, ',\1,')
-      f.flush
+    output_to_write = output_to_write.gsub(/,\s*["|'](\d+)["|']\s*\]/, ',\1]')
-      `cat #{f.path} | yj -jt > #{@config_filename}`
+    File.write(@config_filename, output_to_write)
-    end
   end
 
   def current_config
-    JSON.parse(`cat #{@config_filename} | yj -tj`)
+    loadable_string = File.read(@config_filename)
-  end
+    loadable_string = loadable_string.gsub(/,\s*(\d+)\s*,/,  ', "\1",')
-
+    loadable_string = loadable_string.gsub(/,\s*(\d+)\s*\]/, ', "\1"]')
-  def raw_config_file
+    TOML.load(loadable_string)
-    File.read(@config_filename)
   end
 
   def reload_config
-    conn = PG.connect(admin_connection_string)
+    `kill -s HUP #{@pid}`
-
+    sleep 0.5
-    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
@@ -126,16 +112,10 @@ class PgcatProcess
     "postgresql://#{username}:#{password}@0.0.0.0:#{@port}/pgcat"
   end
 
-  def connection_string(pool_name, username, password = nil, parameters: {})
+  def connection_string(pool_name, username, password = nil)
     cfg = current_config
     user_idx, user_obj = cfg["pools"][pool_name]["users"].detect { |k, user| user["username"] == username }
-    connection_string = "postgresql://#{username}:#{password || user_obj["password"]}@0.0.0.0:#{@port}/#{pool_name}"
+    "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_i, 0],
+      [mirror_host, mirror_pg.port.to_s, "0"],
-      [mirror_host, mirror_pg.port.to_i, 0],
+      [mirror_host, mirror_pg.port.to_s, "0"],
-      [mirror_host, mirror_pg.port.to_i, 0],
+      [mirror_host, mirror_pg.port.to_s, "0"],
     ]
     processes.pgcat.update_config(new_configs)
     processes.pgcat.reload_config
@@ -31,8 +31,7 @@ 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)
-    # Allow some slack in mirroring successes
+    expect(mirror_pg.count_select_1_plus_2).to eq(runs * 3)
-    expect(mirror_pg.count_select_1_plus_2).to be > ((runs - 5) * 3)
   end
 
   context "when main server connection is closed" do
@@ -43,9 +42,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_i, 0],
+          [mirror_host, mirror_pg.port.to_s, "0"],
-          [mirror_host, mirror_pg.port.to_i, 0],
+          [mirror_host, mirror_pg.port.to_s, "0"],
-          [mirror_host, mirror_pg.port.to_i, 0],
+          [mirror_host, mirror_pg.port.to_s, "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 RESET ALL unless necessary" do
+      it "Does not send DISCARD 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("RESET ALL")).to eq(0)
+        expect(processes.primary.count_query("DISCARD 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("RESET ALL")).to eq(10)
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
       end
 
       it "Resets server roles correctly" do
@@ -273,7 +273,7 @@ describe "Miscellaneous" do
         end
       end
 
-      it "Does not send RESET ALL unless necessary" do
+      it "Does not send DISCARD 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("RESET ALL")).to eq(0)
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
 
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
@@ -292,32 +292,8 @@ describe "Miscellaneous" do
           conn.close
         end
 
-        expect(processes.primary.count_query("RESET ALL")).to eq(10)
+        expect(processes.primary.count_query("DISCARD 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
@@ -331,7 +307,7 @@ describe "Miscellaneous" do
           conn.async_exec("COMMIT")
           conn.close
         end
-        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
 
         10.times do
           conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
@@ -341,7 +317,7 @@ describe "Miscellaneous" do
           conn.async_exec("COMMIT")
           conn.close
         end
-        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
       end
     end
 
@@ -354,7 +330,8 @@ describe "Miscellaneous" do
         conn.async_exec("SET statement_timeout TO 1000")
         conn.close
 
-        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+        puts processes.pgcat.logs
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
       end
 
       it "will not clean up prepared statements" do
@@ -364,7 +341,8 @@ describe "Miscellaneous" do
 
         conn.close
 
-        expect(processes.primary.count_query("RESET ALL")).to eq(0)
+        puts processes.pgcat.logs
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
       end
     end
   end
@@ -374,6 +352,7 @@ 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
   

tests/ruby/prepared_spec.rb

@@ -1,214 +1,29 @@
 require_relative 'spec_helper'
 
 describe 'Prepared statements' do
-  let(:pool_size) { 5 }
+  let(:processes) { Helpers::Pgcat.three_shard_setup('sharded_db', 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
+  context 'enabled' do
-    new_configs = processes.pgcat.current_config
+    it 'will work over the same connection' do
-    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])
+        conn.describe_prepared(statement_name)
       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
+    it 'will work with new connections' do
-      10.times do |i|
+      10.times do
         conn = PG.connect(processes.pgcat.connection_string('sharded_db', 'sharding_user'))
-        statement_name = "statement_#{i}"
+
-        conn.prepare(statement_name, 'SELECT $1::int')
+        statement_name = 'statement1'
-        conn.exec_prepared(statement_name, [1])
+        conn.prepare('statement1', 'SELECT $1::int')
+        conn.exec_prepared('statement1', [1])
+        conn.describe_prepared('statement1')
       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,11 +19,10 @@ 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
@@ -49,148 +48,4 @@ 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

@@ -233,19 +233,17 @@ describe "Stats" do
         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|
-        %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["maxwait"]).to eq("1")
         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 maxwait].each do |s|
+        %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")
@@ -257,24 +255,23 @@ describe "Stats" do
 
       it "show correct max_wait" do
         threads = []
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
         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
-        sleep(1.1)
-        results = admin_conn.async_exec("SHOW POOLS")[0]
-        # Value is only reported when there are clients waiting
-        expect(results["maxwait"]).to eq("1")
-        expect(results["maxwait_us"].to_i).to be_within(20_000).of(100_000)
 
         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]
-        # no clients are waiting so value is 0
+
-        expect(results["maxwait"]).to eq("0")
+        expect(results["maxwait"]).to eq("1")
-        expect(results["maxwait_us"]).to eq("0")
+        expect(results["maxwait_us"].to_i).to be_within(200_000).of(500_000)
         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
@@ -332,40 +329,6 @@ describe "Stats" do
       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
 
 

tests/rust/src/main.rs

@@ -15,13 +15,8 @@ async fn test_prepared_statements() {
     for _ in 0..5 {
         let pool = pool.clone();
         let handle = tokio::task::spawn(async move {
-            for i in 0..1000 {
+            for _ in 0..1000 {
-                match sqlx::query(&format!("SELECT {:?}", i % 5)).fetch_all(&pool).await {
+                sqlx::query("SELECT 1").fetch_all(&pool).await.unwrap();
-                    Ok(_) => (),
-                    Err(err) => {
-                        panic!("prepared statement error: {}", err);
-                    }
-                }
             }
         });
 

utilities/deb.sh

@@ -1,40 +0,0 @@
-#!/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