Improve row decoding performance (#221)

* Leverage better-performing new APIs from PostgresNIO, also solving (most) deprecation warnings. Also some additional minor improvements to connection handling.

* Generating NIOSSLContexts is (very) expensive, do it only once (per connection source, at least) rather than for every connection.

* CI cleanup - update Swift versions, use generic names, don't over-test, use new checkout action

.github/workflows/test.yml

@@ -9,84 +9,86 @@ jobs:
         dbimage:
           - postgres:14
           - postgres:13
-          - postgres:12
           - postgres:11
         dbauth:
           - trust
           - md5
           - scram-sha-256
         swiftver:
-          - 'swift:5.2'
-          - 'swift:5.5'
-          - 'swiftlang/swift:nightly-main'
+          - swift:5.2
+          - swift:5.5
+          - swift:5.6
+          - swiftlang/swift:nightly-main
         swiftos:
           - focal
+        include:
+          - swiftver: swift:5.2
+            test_flag: --enable-test-discovery
     container: ${{ format('{0}-{1}', matrix.swiftver, matrix.swiftos) }}
     runs-on: ubuntu-latest
     env:
       LOG_LEVEL: debug
       POSTGRES_HOSTNAME: 'psql-a'
       POSTGRES_HOSTNAME_A: 'psql-a'
       POSTGRES_HOSTNAME_B: 'psql-b'
-      POSTGRES_DB: 'vapor_database'
-      POSTGRES_DB_A: 'vapor_database'
-      POSTGRES_DB_B: 'vapor_database'
-      POSTGRES_USER: 'vapor_username'
-      POSTGRES_USER_A: 'vapor_username'
-      POSTGRES_USER_B: 'vapor_username'
-      POSTGRES_PASSWORD: 'vapor_password'
-      POSTGRES_PASSWORD_A: 'vapor_password'
-      POSTGRES_PASSWORD_B: 'vapor_password'
+      POSTGRES_DB: 'test_database'
+      POSTGRES_DB_A: 'test_database'
+      POSTGRES_DB_B: 'test_database'
+      POSTGRES_USER: 'test_username'
+      POSTGRES_USER_A: 'test_username'
+      POSTGRES_USER_B: 'test_username'
+      POSTGRES_PASSWORD: 'test_password'
+      POSTGRES_PASSWORD_A: 'test_password'
+      POSTGRES_PASSWORD_B: 'test_password'
     services:
       psql-a:
         image: ${{ matrix.dbimage }}
         env:
-          POSTGRES_USER: 'vapor_username'
-          POSTGRES_DB: 'vapor_database'
-          POSTGRES_PASSWORD: 'vapor_password'
+          POSTGRES_USER: 'test_username'
+          POSTGRES_DB: 'test_database'
+          POSTGRES_PASSWORD: 'test_password'
           POSTGRES_HOST_AUTH_METHOD: ${{ matrix.dbauth }}
           POSTGRES_INITDB_ARGS: --auth-host=${{ matrix.dbauth }}
       psql-b:
         image: ${{ matrix.dbimage }}
         env:
-          POSTGRES_USER: 'vapor_username'
-          POSTGRES_DB: 'vapor_database'
-          POSTGRES_PASSWORD: 'vapor_password'
+          POSTGRES_USER: 'test_username'
+          POSTGRES_DB: 'test_database'
+          POSTGRES_PASSWORD: 'test_password'
           POSTGRES_HOST_AUTH_METHOD: ${{ matrix.dbauth }}
           POSTGRES_INITDB_ARGS: --auth-host=${{ matrix.dbauth }}
     steps:
       - name: Check out package
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
         with: { path: 'postgres-kit' }
       - name: Check out fluent-postgres-driver dependent
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
         with: { repository: 'vapor/fluent-postgres-driver', path: 'fluent-postgres-driver' }
       - name: Run local tests with Thread Sanitizer
-        run: swift test --package-path postgres-kit --enable-test-discovery --sanitize=thread
+        run: swift test --package-path postgres-kit ${{ matrix.test_flag }} --sanitize=thread
       - name: Use local package
         run: swift package --package-path fluent-postgres-driver edit postgres-kit --path postgres-kit
       - name: Run fluent-postgres-kit tests with Thread Sanitizer
-        run: swift test --package-path fluent-postgres-driver --enable-test-discovery --sanitize=thread
+        run: swift test --package-path fluent-postgres-driver ${{ matrix.test_flag }} --sanitize=thread
 
   macos:
     strategy:
       fail-fast: false
       matrix:
+        # Only test latest version and one auth method on macOS
         dbimage:
-          # Only test the lastest version on macOS, let Linux do the rest
           - postgresql@14
         dbauth:
-          # Only test one auth method on macOS, Linux tests will cover the others
           - scram-sha-256
         xcode:
           - latest-stable
-          - latest
+          #- latest
     runs-on: macos-11
     env:
       LOG_LEVEL: debug
       POSTGRES_HOSTNAME: 127.0.0.1
-      POSTGRES_USER: vapor_username
-      POSTGRES_PASSWORD: vapor_password
+      POSTGRES_USER: test_username
+      POSTGRES_PASSWORD: test_password
       POSTGRES_DB: postgres
       POSTGRES_HOST_AUTH_METHOD: ${{ matrix.dbauth }}
     steps:
@@ -102,7 +104,7 @@ jobs:
           pg_ctl start --wait
         timeout-minutes: 2
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Run local tests with Thread Sanitizer
         run: |
           swift test --sanitize=thread -Xlinker -rpath \

Package.swift

@@ -10,7 +10,7 @@ let package = Package(
         .library(name: "PostgresKit", targets: ["PostgresKit"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.7.2"),
+        .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.9.0"),
         .package(url: "https://github.com/vapor/sql-kit.git", from: "3.16.0"),
         .package(url: "https://github.com/vapor/async-kit.git", from: "1.0.0"),
     ],

Sources/PostgresKit/PostgresConfiguration.swift

@@ -15,6 +15,7 @@ public struct PostgresConfiguration {
     public static var ianaPortNumber: Int { 5432 }
 
     internal var _hostname: String?
+    internal var _port: Int?
 
     public init?(url: String) {
         guard let url = URL(string: url) else {
@@ -67,6 +68,7 @@ public struct PostgresConfiguration {
         self.database = database
         self.tlsConfiguration = nil
         self._hostname = nil
+        self._port = nil
     }
 
     public init(
@@ -85,5 +87,6 @@ public struct PostgresConfiguration {
         self.password = password
         self.tlsConfiguration = tlsConfiguration
         self._hostname = hostname
+        self._port = port
     }
 }

Sources/PostgresKit/PostgresConnectionSource.swift

@@ -1,44 +1,79 @@
+import NIOConcurrencyHelpers
+import NIOSSL
+
 public struct PostgresConnectionSource: ConnectionPoolSource {
     public let configuration: PostgresConfiguration
+    public let sslContext: Result<NIOSSLContext?, Error>
+    private static let idGenerator = NIOAtomic.makeAtomic(value: 0)
 
     public init(configuration: PostgresConfiguration) {
         self.configuration = configuration
+        // TODO: Figure out a way to throw errors from this initializer sensibly, or to lazily init the NIOSSLContext only once in makeConnection()
+        self.sslContext = .init(catching: { try configuration._hostname.flatMap { _ in try configuration.tlsConfiguration.map { try .init(configuration: $0) } } })
     }
 
     public func makeConnection(
         logger: Logger,
         on eventLoop: EventLoop
     ) -> EventLoopFuture<PostgresConnection> {
-        let address: SocketAddress
-        do {
-            address = try self.configuration.address()
-        } catch {
-            return eventLoop.makeFailedFuture(error)
-        }
-        return PostgresConnection.connect(
-            to: address,
-            tlsConfiguration: self.configuration.tlsConfiguration,
-            serverHostname: self.configuration._hostname,
-            logger: logger,
-            on: eventLoop
-        ).flatMap { conn in
-            return conn.authenticate(
-                username: self.configuration.username,
-                database: self.configuration.database,
-                password: self.configuration.password,
+        if let hostname = self.configuration._hostname {
+            let tlsMode: PostgresConnection.Configuration.TLS
+            switch self.sslContext {
+                case let .success(sslContext): tlsMode = sslContext.map { .require($0) } ?? .disable
+                case let .failure(error): return eventLoop.makeFailedFuture(error)
+            }
+            let future = PostgresConnection.connect(
+                on: eventLoop,
+                configuration: .init(
+                    connection: .init(host: hostname, port: self.configuration._port ?? PostgresConfiguration.ianaPortNumber),
+                    authentication: .init(username: self.configuration.username, database: self.configuration.database, password: self.configuration.password),
+                    tls: tlsMode
+                ),
+                id: Self.idGenerator.add(1),
                 logger: logger
-            ).flatMap {
-                if let searchPath = self.configuration.searchPath {
-                    let string = searchPath.map { "\"" + $0 + "\"" }.joined(separator: ", ")
-                    return conn.simpleQuery("SET search_path = \(string)")
-                        .map { _ in }
-                } else {
-                    return eventLoop.makeSucceededFuture(())
+            )
+
+            if let searchPath = self.configuration.searchPath {
+                return future.flatMap { conn in
+                    let string = searchPath.map { #""\#($0)""# }.joined(separator: ", ")
+                    return conn.simpleQuery("SET search_path = \(string)").map { _ in conn }
+                }
+            } else {
+                return future
+            }
+        } else {
+            let address: SocketAddress
+            do {
+                address = try self.configuration.address()
+            } catch {
+                return eventLoop.makeFailedFuture(error)
+            }
+
+            // Legacy code path until PostgresNIO regains support for connecting directly to a SocketAddress.
+            return PostgresConnection.connect(
+                to: address,
+                tlsConfiguration: self.configuration.tlsConfiguration,
+                serverHostname: self.configuration._hostname,
+                logger: logger,
+                on: eventLoop
+            ).flatMap { conn in
+                return conn.authenticate(
+                    username: self.configuration.username,
+                    database: self.configuration.database,
+                    password: self.configuration.password,
+                    logger: logger
+                ).flatMap {
+                    if let searchPath = self.configuration.searchPath {
+                        let string = searchPath.map { "\"" + $0 + "\"" }.joined(separator: ", ")
+                        return conn.simpleQuery("SET search_path = \(string)").map { _ in conn }
+                    } else {
+                        return eventLoop.makeSucceededFuture(conn)
+                    }
+                }.flatMapErrorThrowing { error in
+                    _ = conn.close()
+                    throw error
                 }
-            }.flatMapErrorThrowing { error in
-                _ = conn.close()
-                throw error
-            }.map { conn }
+            }
         }
     }
 }

Sources/PostgresKit/PostgresRow+SQL.swift

@@ -1,36 +1,40 @@
 extension PostgresRow {
     public func sql(decoder: PostgresDataDecoder = .init()) -> SQLRow {
-        return _PostgreSQLRow(row: self, decoder: decoder)
+        return _PostgresSQLRow(row: self.makeRandomAccess(), decoder: decoder)
     }
 }
 
 // MARK: Private
 
-private struct _PostgreSQLRow: SQLRow {
-    let row: PostgresRow
+private struct _PostgresSQLRow: SQLRow {
+    let randomAccessView: PostgresRandomAccessRow
     let decoder: PostgresDataDecoder
 
     enum _Error: Error {
         case missingColumn(String)
     }
+
+    init(row: PostgresRandomAccessRow, decoder: PostgresDataDecoder) {
+        self.randomAccessView = row
+        self.decoder = decoder
+    }
 
     var allColumns: [String] {
-        self.row.rowDescription.fields.map { $0.name }
+        self.randomAccessView.map { $0.columnName }
     }
 
     func contains(column: String) -> Bool {
-        self.row.rowDescription.fields
-            .contains { $0.name == column }
+        self.randomAccessView.contains(column)
     }
 
     func decodeNil(column: String) throws -> Bool {
-        self.row.column(column)?.value == nil
+        !self.randomAccessView.contains(column) || self.randomAccessView[column].bytes == nil
     }
 
     func decode<D>(column: String, as type: D.Type) throws -> D where D : Decodable {
-        guard let data = self.row.column(column) else {
+        guard self.randomAccessView.contains(column) else {
             throw _Error.missingColumn(column)
         }
-        return try self.decoder.decode(D.self, from: data)
+        return try self.decoder.decode(D.self, from: self.randomAccessView[data: column])
     }
 }

Tests/PostgresKitTests/Utilities.swift

@@ -1,6 +1,5 @@
 import XCTest
 import PostgresKit
-import NIOCore
 import Logging
 #if canImport(Darwin)
 import Darwin.C
@@ -10,28 +9,7 @@ import Glibc
 
 extension PostgresConnection {
     static func test(on eventLoop: EventLoop) -> EventLoopFuture<PostgresConnection> {
-        let config = PostgresConfiguration.test
-
-        return eventLoop.flatSubmit { () -> EventLoopFuture<PostgresConnection> in
-            do {
-                let address = try config.address()
-                return self.connect(to: address, on: eventLoop)
-            } catch {
-                return eventLoop.makeFailedFuture(error)
-            }
-        }.flatMap { conn in
-            return conn.authenticate(
-                username: config.username,
-                database: config.database,
-                password: config.password
-            )
-            .map { conn }
-            .flatMapError { error in
-                conn.close().flatMapThrowing {
-                    throw error
-                }
-            }
-        }
+        return PostgresConnectionSource(configuration: .test).makeConnection(logger: .init(label: "vapor.codes.postgres-kit.test"), on: eventLoop)
     }
 }