feat(postgresql): show server notices with each statement's result

apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleCoreConnection.swift

@@ -105,6 +105,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
         var queryTimeoutSeconds = 0
         var sessionSchema: String?
         var capturesServerOutput = false
+        var pendingServerOutput = OracleServerOutputPending()
         var close = OracleCloseRecord()
         var serverRelease: OracleServerRelease?
         var loginWarning: String?
@@ -635,6 +636,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
         await queryGate.acquire()
         do {
             let connection = try requireConnection()
+            noteStatement()
             let rows = try await withQueryDeadline { [self] in
                 try await runDirect(sql, binds: binds, autoCommit: autoCommit, readsRows: readsRows, on: connection)
             }
@@ -696,14 +698,19 @@ public final class OracleCoreConnection: @unchecked Sendable {
         guard state.withLock({ $0.capturesServerOutput }), maxLines > 0 else { return .empty }
         await queryGate.acquire()
 
-        guard let connection = state.withLock({ $0.isConnected ? $0.nioConnection : nil }) else {
+        let connection = state.withLock { current -> OracleNIO.OracleConnection? in
+            guard current.isConnected, current.pendingServerOutput.needsRead else { return nil }
+            return current.nioConnection
+        }
+        guard let connection else {
             await queryGate.release()
             return .empty
         }
         do {
             let output = try await withQueryDeadline { [self] in
                 try await readServerOutput(on: connection, maxLines: maxLines)
             }
+            state.withLock { $0.pendingServerOutput.noteRead() }
             await queryGate.release()
             return output
         } catch {
@@ -713,6 +720,10 @@ public final class OracleCoreConnection: @unchecked Sendable {
         }
     }
 
+    private func noteStatement() {
+        state.withLock { $0.pendingServerOutput.noteStatement() }
+    }
+
     private func readServerOutput(
         on connection: OracleNIO.OracleConnection,
         maxLines: Int
@@ -827,6 +838,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
 
         do {
             let connection = try await reconnectedConnection()
+            noteStatement()
             let admitted = try admit(role)
             let result = try await withQueryDeadline { [self] in
                 let zones = await sessionZones(on: connection)
@@ -859,6 +871,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
         let admitted: (options: StatementOptions, session: Int)
         do {
             connection = try await reconnectedConnection()
+            noteStatement()
             admitted = try admit(role)
             try await withQueryDeadline { [self] in
                 try await discardRows(OracleStatement(unsafeSQL: statement), options: admitted.options, on: connection)
@@ -983,6 +996,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
 
         do {
             let connection = try await reconnectedConnection()
+            noteStatement()
             let admitted = try admit(role)
             try await withQueryDeadline { [self] in
                 let zones = await sessionZones(on: connection)

apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleServerOutputPending.swift

@@ -0,0 +1,12 @@
+// A reconnect never clears this: a stale true costs one round trip, a stale false loses a statement's DBMS_OUTPUT lines.
+internal struct OracleServerOutputPending: Sendable, Equatable {
+    private(set) var needsRead = false
+
+    mutating func noteStatement() {
+        needsRead = true
+    }
+
+    mutating func noteRead() {
+        needsRead = false
+    }
+}

apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleServerOutputPendingTests.swift

@@ -0,0 +1,39 @@
+@testable import TableProOracleCore
+import Testing
+
+// The host reads DBMS_OUTPUT before and after every editor statement, and on Oracle each read is a round trip.
+struct OracleServerOutputPendingTests {
+    @Test func testANewSessionHasNothingToRead() {
+        #expect(!OracleServerOutputPending().needsRead)
+    }
+
+    @Test func testAStatementLeavesOutputToRead() {
+        var pending = OracleServerOutputPending()
+        pending.noteStatement()
+        #expect(pending.needsRead)
+    }
+
+    @Test func testOneReadTakesWhatEveryStatementBeforeItLeft() {
+        var pending = OracleServerOutputPending()
+        pending.noteStatement()
+        pending.noteStatement()
+        pending.noteRead()
+        #expect(!pending.needsRead)
+    }
+
+    @Test func testAReadBeforeTheNextStatementFindsNothingNew() {
+        var pending = OracleServerOutputPending()
+        pending.noteStatement()
+        pending.noteRead()
+        pending.noteRead()
+        #expect(!pending.needsRead)
+    }
+
+    @Test func testAStatementAfterAReadNeedsAnotherRead() {
+        var pending = OracleServerOutputPending()
+        pending.noteStatement()
+        pending.noteRead()
+        pending.noteStatement()
+        #expect(pending.needsRead)
+    }
+}

apple/Plugins/MSSQLDriverPlugin/FreeTDSConnection.swift

@@ -101,6 +101,14 @@ nonisolated private func freetdsDiagnostics(for dbproc: UnsafeMutablePointer<DBP
     return freetdsErrors.withLock { $0.perConnection[key] ?? FreeTDSDiagnostics() }
 }
 
+nonisolated private func freetdsRestoreDiagnostics(
+    _ diagnostics: FreeTDSDiagnostics,
+    for dbproc: UnsafeMutablePointer<DBPROCESS>
+) {
+    let key = freetdsConnectionKey(dbproc)
+    freetdsErrors.withLock { $0.perConnection[key] = diagnostics }
+}
+
 /// The output of the connection's latest request, handed over once.
 nonisolated private func freetdsTakeOutput(
     for dbproc: UnsafeMutablePointer<DBPROCESS>
@@ -585,7 +593,9 @@ nonisolated final class FreeTDSConnection: @unchecked Sendable {
             answer = .failure(error)
         }
 
-        guard dbproc != nil else { return try answer.get() }
+        guard let proc = dbproc else { return try answer.get() }
+        // The reset clears the session's messages, and the statement's PRINT output is what the next read takes.
+        let statementDiagnostics = freetdsDiagnostics(for: proc)
         // Numbered after the statement, so a Stop that ended it does not skip the reset.
         let resetCall = requestGate.enqueue()
         do {
@@ -597,6 +607,7 @@ nonisolated final class FreeTDSConnection: @unchecked Sendable {
             disconnect()
             throw MSSQLCoreError.queryFailed(MSSQLShowplan.resetFailedMessage)
         }
+        freetdsRestoreDiagnostics(statementDiagnostics, for: proc)
         return try answer.get()
     }
 

apple/Plugins/PostgreSQLDriverPlugin/LibPQDriverCore.swift

@@ -320,6 +320,11 @@ final class LibPQDriverCore: @unchecked Sendable {
         libpqConnection?.observeServerNotices(observer)
     }
 
+    func fetchServerOutput() -> PluginServerOutput {
+        guard let connection = libpqConnection else { return .none }
+        return connection.takeServerOutput()
+    }
+
     func nextNotifications(waitMicroseconds: Int64) async throws -> [LibPQNotification] {
         guard let connection = libpqConnection else { throw LibPQPluginError.notConnected }
         return try await connection.nextNotifications(waitMicroseconds: waitMicroseconds)
@@ -548,6 +553,10 @@ extension LibPQBackedDriver {
         await core.sessionTransactionState()
     }
 
+    func fetchServerOutput() async throws -> PluginServerOutput {
+        core.fetchServerOutput()
+    }
+
     func switchSchema(to schema: String) async throws {
         try await core.applySchema(schema)
     }

apple/Plugins/PostgreSQLDriverPlugin/LibPQNoticeOutput.swift

@@ -0,0 +1,62 @@
+//
+//  LibPQNoticeOutput.swift
+//  PostgreSQLDriverPlugin
+//
+
+import Foundation
+import TableProPluginKit
+
+struct LibPQNoticeOutput: Sendable {
+    static let lineLimit = 10_000
+    static let lineByteLimit = 32_767
+
+    private var lines: [String] = []
+    private var isTruncated = false
+
+    // libpq's own notices carry no SQLSTATE, and a session-ending message is reported as the lost connection instead.
+    static func isOutput(severity: String?, sqlState: String?) -> Bool {
+        guard let sqlState, !sqlState.isEmpty else { return false }
+        return !LibPQServerMessage.endsSession(severity: severity, sqlState: sqlState)
+    }
+
+    // Scalars, not Characters: "\r\n" is one Character, so a message ending in "\r" would keep libpq's newline.
+    mutating func record(_ text: String) {
+        guard hasRoom() else { return }
+        var body = Substring(text).unicodeScalars
+        if body.last == "\n" { body.removeLast() }
+        guard !body.isEmpty else { return }
+        // One line at a time: a notice can hold millions of newlines, and only the first ones fit.
+        var start = body.startIndex
+        while hasRoom() {
+            let end = body[start...].firstIndex(of: "\n") ?? body.endIndex
+            lines.append(clipped(body[start..<end]))
+            guard end < body.endIndex else { return }
+            start = body.index(after: end)
+        }
+    }
+
+    mutating func take() -> PluginServerOutput {
+        let output = PluginServerOutput(lines: lines, isTruncated: isTruncated)
+        self = Self()
+        return output
+    }
+
+    private mutating func hasRoom() -> Bool {
+        guard lines.count < Self.lineLimit else {
+            isTruncated = true
+            return false
+        }
+        return true
+    }
+
+    private mutating func clipped(_ line: Substring.UnicodeScalarView) -> String {
+        let bytes = Substring(line).utf8
+        guard bytes.count > Self.lineByteLimit else { return String(Substring(line)) }
+        isTruncated = true
+        var end = bytes.index(bytes.startIndex, offsetBy: Self.lineByteLimit)
+        while UTF8.isContinuation(bytes[end]) {
+            bytes.formIndex(before: &end)
+        }
+        return String(Substring(line[..<end]))
+    }
+}

apple/Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift

@@ -1049,7 +1049,7 @@ final class LibPQPluginConnection: @unchecked Sendable {
         /// and the connection is drained instead.
         let cancelsOutput = cancelsAbandonedOutput(conn)
         let suppressCancel = !cancelsOutput
-        _ = finishPendingResults(conn, cancellingOutput: cancelsOutput)
+        finishAbandonedResults(conn, cancellingOutput: cancelsOutput)
         if let ended = sessionEndedBeforeSending(conn) { throw ended }
 
         let localQuery = String(query)
@@ -1366,7 +1366,7 @@ final class LibPQPluginConnection: @unchecked Sendable {
                 /// Read before the query goes out: once it is in flight the status is
                 /// PQTRANS_ACTIVE, and the transaction this guard exists for is invisible.
                 let cancelsOutput = cancelsAbandonedOutput(conn)
-                _ = finishPendingResults(conn, cancellingOutput: cancelsOutput)
+                finishAbandonedResults(conn, cancellingOutput: cancelsOutput)
                 if let ended = sessionEndedBeforeSending(conn) {
                     continuation.finish(throwing: ended)
                     return
@@ -1602,6 +1602,20 @@ extension LibPQPluginConnection {
         sink?.observeNotices(observer)
     }
 
+    func takeServerOutput() -> PluginServerOutput {
+        guard let sink = stateLock.withLock({ serverMessages?.takeUnretainedValue() }) else { return .none }
+        return sink.takeOutput()
+    }
+
+    // Notices from a result an earlier statement left unread would otherwise show as this statement's output.
+    private func finishAbandonedResults(_ conn: OpaquePointer, cancellingOutput: Bool) {
+        guard let sink = stateLock.withLock({ serverMessages?.takeUnretainedValue() }) else {
+            _ = finishPendingResults(conn, cancellingOutput: cancellingOutput)
+            return
+        }
+        sink.discardingOutput { _ = finishPendingResults(conn, cancellingOutput: cancellingOutput) }
+    }
+
     // A debug session runs with no statement timeout, so a call blocked on a server that stopped answering would
     // otherwise keep the socket, and the PQfinish queued behind it, for as long as TCP keeps the connection open.
     func interruptBlockedCallOnDisconnect() {

apple/Plugins/PostgreSQLDriverPlugin/LibPQServerMessageSink.swift

@@ -5,6 +5,7 @@
 
 import CLibPQ
 import Foundation
+import TableProPluginKit
 
 struct LibPQServerNotice: Sendable {
     let message: String
@@ -19,6 +20,8 @@ final class LibPQServerMessageSink: @unchecked Sendable {
     private var sessionEnding: LibPQPluginError?
     private var forwardedReceiver: PQnoticeReceiver?
     private var noticeObserver: (@Sendable (LibPQServerNotice) -> Void)?
+    private var output = LibPQNoticeOutput()
+    private var discardsOutput = false
 
     static func install(on connection: OpaquePointer) -> Unmanaged<LibPQServerMessageSink> {
         let retained = Unmanaged.passRetained(LibPQServerMessageSink())
@@ -33,10 +36,7 @@ final class LibPQServerMessageSink: @unchecked Sendable {
         lock.withLock { sessionEnding }
     }
 
-    /// Only ever called for a connection libpq still reports as `CONNECTION_OK`. A server before
-    /// 9.6 sends no non-localized severity, so a `RAISE WARNING ... ERRCODE '08006'` on a healthy
-    /// session reaches `receive` looking exactly like the real thing; a session that ended cannot
-    /// be healthy, so the healthy case is the one that must not keep the message.
+    // Before 9.6 the server sends no V field, so a healthy session's `RAISE WARNING ... ERRCODE '08006'` looks session-ending.
     func clearIfHealthy() {
         lock.withLock { sessionEnding = nil }
     }
@@ -47,18 +47,32 @@ final class LibPQServerMessageSink: @unchecked Sendable {
         lock.withLock { noticeObserver = observer }
     }
 
+    func takeOutput() -> PluginServerOutput {
+        lock.withLock { output.take() }
+    }
+
+    func discardingOutput<Value>(_ body: () throws -> Value) rethrows -> Value {
+        lock.withLock { discardsOutput = true }
+        defer { lock.withLock { discardsOutput = false } }
+        return try body()
+    }
+
     fileprivate func receive(_ result: OpaquePointer) {
-        let message = PQresultErrorMessage(result).map {
-            String(cString: $0).trimmingCharacters(in: .whitespacesAndNewlines)
-        } ?? ""
+        let text = PQresultErrorMessage(result).map { String(cString: $0) } ?? ""
+        let message = text.trimmingCharacters(in: .whitespacesAndNewlines)
         let received = LibPQPluginError(message: message) { field in
             PQresultErrorField(result, field).map { String(cString: $0) }
         }
         let severity = PQresultErrorField(result, Self.severityField).map { String(cString: $0) }
+        let isOutput = LibPQNoticeOutput.isOutput(severity: severity, sqlState: received.sqlState)
         let (forward, observer) = lock.withLock { () -> (PQnoticeReceiver?, (@Sendable (LibPQServerNotice) -> Void)?) in
             if LibPQServerMessage.endsSession(severity: severity, sqlState: received.sqlState) {
                 sessionEnding = received
             }
+            // The debugger shows what its observer gets, and nothing reads output from its target session.
+            if isOutput, noticeObserver == nil, !discardsOutput {
+                output.record(text)
+            }
             return (forwardedReceiver, noticeObserver)
         }
         if let observer {