Merge pull request #64 from haskellari/pipeline

Add pipeline mode API

Author: phadej

.github/workflows/haskell-ci.yml

@@ -94,6 +94,8 @@ jobs:
           chmod a+x "$HOME/.ghcup/bin/ghcup"
           "$HOME/.ghcup/bin/ghcup" install ghc "$HCVER" || (cat "$HOME"/.ghcup/logs/*.* && false)
           "$HOME/.ghcup/bin/ghcup" install cabal 3.12.1.0 || (cat "$HOME"/.ghcup/logs/*.* && false)
+          apt-get update
+          apt-get install -y libpq-dev
         env:
           HCKIND: ${{ matrix.compilerKind }}
           HCNAME: ${{ matrix.compiler }}
@@ -233,7 +235,7 @@ jobs:
           $CABAL v2-build $ARG_COMPILER $ARG_TESTS $ARG_BENCH all --write-ghc-environment-files=always
       - name: tests
         run: |
-          $CABAL v2-test $ARG_COMPILER $ARG_TESTS $ARG_BENCH all
+          $CABAL v2-test $ARG_COMPILER $ARG_TESTS $ARG_BENCH all --test-show-details=direct
       - name: cabal check
         run: |
           cd ${PKGDIR_postgresql_libpq} || false
@@ -244,7 +246,7 @@ jobs:
           ${CABAL} -vnormal check
       - name: haddock
         run: |
-          $CABAL v2-haddock --disable-documentation $ARG_COMPILER --with-haddock $HADDOCK $ARG_TESTS $ARG_BENCH all
+          $CABAL v2-haddock --disable-documentation --haddock-all $ARG_COMPILER --with-haddock $HADDOCK $ARG_TESTS $ARG_BENCH all
       - name: unconstrained build
         run: |
           rm -f cabal.project.local

.github/workflows/simple.yml

@@ -39,6 +39,7 @@ jobs:
           username: ci
           password: sw0rdfish
           database: test
+          postgres-version: "14"
 
       - name: Checkout
         uses: actions/checkout@v4

cabal.haskell-ci

@@ -1,9 +1,6 @@
 branches:           master
 postgresql:         True
-
--- due build-type: Custom
-test-output-direct: False
-haddock-components: libs
+apt:                libpq-dev
 
 constraint-set pkg-config
   constraints: postgresql-libpq +use-pkg-config

postgresql-libpq-configure/configure

@@ -3570,7 +3570,7 @@ then :
   ac_cv_POSTGRESQL_LIBS="$POSTGRESQL_LIBS"
 fi
 
-    postgresql_version_req=10.22
+    postgresql_version_req=14.12
     found_postgresql="no"
 
     POSTGRESQL_VERSION=""

postgresql-libpq-configure/configure.ac

@@ -5,7 +5,7 @@ AC_CONFIG_MACRO_DIRS([m4])
 AC_PROG_CC
 AC_PROG_SED
 
-AX_LIB_POSTGRESQL([10.22])
+AX_LIB_POSTGRESQL([14.12])
 
 POSTGRESQL_EXTRA_LIBS="pq"
 POSTGRESQL_LIBDIR=$(echo "$POSTGRESQL_LDFLAGS"|$SED 's/-L//')

postgresql-libpq-configure/postgresql-libpq-configure.cabal

@@ -1,6 +1,6 @@
 cabal-version:      2.4
 name:               postgresql-libpq-configure
-version:            0.10.0.1
+version:            0.11
 synopsis:           low-level binding to libpq: configure based provider
 description:
   This is a binding to libpq: the C application

postgresql-libpq-pkgconfig/postgresql-libpq-pkgconfig.cabal

@@ -1,6 +1,6 @@
 cabal-version:      2.4
 name:               postgresql-libpq-pkgconfig
-version:            0.10
+version:            0.11
 synopsis:           low-level binding to libpq: pkg-config based provider
 description:
   This is a binding to libpq: the C application
@@ -35,7 +35,7 @@ extra-source-files: CHANGELOG.md
 library
   default-language:  Haskell2010
   build-depends:     base <5
-  pkgconfig-depends: libpq >=10.22
+  pkgconfig-depends: libpq >=14.12
 
 source-repository head
   type:     git

postgresql-libpq.cabal

@@ -1,6 +1,6 @@
 cabal-version:      2.4
 name:               postgresql-libpq
-version:            0.10.2.0
+version:            0.11.0.0
 synopsis:           low-level binding to libpq
 description:
   This is a binding to libpq: the C application
@@ -81,10 +81,10 @@ library
     build-depends: Win32 >=2.2.0.2 && <2.15
 
   if flag(use-pkg-config)
-    build-depends: postgresql-libpq-pkgconfig ^>=0.10
+    build-depends: postgresql-libpq-pkgconfig ^>=0.11
 
   else
-    build-depends: postgresql-libpq-configure ^>=0.10
+    build-depends: postgresql-libpq-configure ^>=0.11
 
   build-tool-depends: hsc2hs:hsc2hs >=0.68.5
 

src/Database/PostgreSQL/LibPQ.hs

@@ -171,6 +171,15 @@ module Database.PostgreSQL.LibPQ
     , FlushStatus(..)
     , flush
 
+    -- * Pipeline Mode
+    -- $pipelinemode
+    , PipelineStatus(..)
+    , pipelineStatus
+    , enterPipelineMode
+    , exitPipelineMode
+    , pipelineSync
+    , sendFlushRequest
+
     -- * Cancelling Queries in Progress
     -- $cancel
     , Cancel
@@ -1640,6 +1649,60 @@ flush connection =
          1 -> return FlushWriting
          _ -> return FlushFailed
 
+-- $pipelinemode
+-- These functions control behaviour in pipeline mode.
+--
+-- Pipeline mode allows applications to send a query
+-- without having to read the result of the previously
+-- sent query. Taking advantage of the pipeline mode,
+-- a client will wait less for the server, since multiple
+-- queries/results can be sent/received in
+-- a single network transaction.
+
+-- | Returns the current pipeline mode status of the libpq connection.
+--
+-- @since 0.11.0.0
+pipelineStatus :: Connection
+               -> IO PipelineStatus
+pipelineStatus connection = do
+    stat <- withConn connection c_PQpipelineStatus
+    maybe
+      (fail $ "Unknown pipeline status " ++ show stat)
+      return
+      (fromCInt stat)
+
+-- | Causes a connection to enter pipeline mode if it is currently idle or already in pipeline mode.
+--
+-- @since 0.11.0.0
+enterPipelineMode :: Connection
+                  -> IO Bool
+enterPipelineMode connection =
+    enumFromConn connection c_PQenterPipelineMode
+
+-- | Causes a connection to exit pipeline mode if it is currently in pipeline mode with an empty queue and no pending results.
+--
+-- @since 0.11.0.0
+exitPipelineMode :: Connection
+                 -> IO Bool
+exitPipelineMode connection =
+    enumFromConn connection c_PQexitPipelineMode
+
+-- | Marks a synchronization point in a pipeline by sending a sync message and flushing the send buffer. This serves as the delimiter of an implicit transaction and an error recovery point>
+--
+-- @since 0.11.0.0
+pipelineSync :: Connection
+             -> IO Bool
+pipelineSync connection =
+    enumFromConn connection c_PQpipelineSync
+
+-- | Sends a request for the server to flush its output buffer.
+--
+-- @since 0.11.0.0
+sendFlushRequest :: Connection
+                 -> IO Bool
+sendFlushRequest connection =
+    enumFromConn connection c_PQsendFlushRequest
+
 
 -- $cancel
 -- A client application can request cancellation of a command that is

src/Database/PostgreSQL/LibPQ/Enums.hsc

@@ -37,23 +37,42 @@ data ExecStatus
     | NonfatalError -- ^ A nonfatal error (a notice or
                     -- warning) occurred.
     | FatalError    -- ^ A fatal error occurred.
-    | SingleTuple   -- ^ The PGresult contains a single result tuple
+    | SingleTuple   -- ^ The 'Result' contains a single result tuple
                     -- from the current command. This status occurs
                     -- only when single-row mode has been selected
                     -- for the query.
+
+    | PipelineSync  -- ^ The 'Result' represents a synchronization
+                    -- point in pipeline mode, requested by
+                    -- 'pipelineSync'. This status occurs only
+                    -- when pipeline mode has been selected.
+                    --
+                    -- @since 0.11.0.0
+
+    | PipelineAbort -- ^ The 'Result' represents a pipeline that
+                    -- has received an error from the server.
+                    -- 'getResult' must be called repeatedly,
+                    -- and each time it will return this status
+                    -- code until the end of the current pipeline,
+                    -- at which point it will return 'PipelineSync'
+                    -- and normal processing can resume.
+                    --
+                    -- @since 0.11.0.0
   deriving (Eq, Show)
 
 instance FromCInt ExecStatus where
-    fromCInt (#const PGRES_EMPTY_QUERY)    = Just EmptyQuery
-    fromCInt (#const PGRES_COMMAND_OK)     = Just CommandOk
-    fromCInt (#const PGRES_TUPLES_OK)      = Just TuplesOk
-    fromCInt (#const PGRES_COPY_OUT)       = Just CopyOut
-    fromCInt (#const PGRES_COPY_IN)        = Just CopyIn
-    fromCInt (#const PGRES_COPY_BOTH)      = Just CopyBoth
-    fromCInt (#const PGRES_BAD_RESPONSE)   = Just BadResponse
-    fromCInt (#const PGRES_NONFATAL_ERROR) = Just NonfatalError
-    fromCInt (#const PGRES_FATAL_ERROR)    = Just FatalError
-    fromCInt (#const PGRES_SINGLE_TUPLE)   = Just SingleTuple
+    fromCInt (#const PGRES_EMPTY_QUERY)      = Just EmptyQuery
+    fromCInt (#const PGRES_COMMAND_OK)       = Just CommandOk
+    fromCInt (#const PGRES_TUPLES_OK)        = Just TuplesOk
+    fromCInt (#const PGRES_COPY_OUT)         = Just CopyOut
+    fromCInt (#const PGRES_COPY_IN)          = Just CopyIn
+    fromCInt (#const PGRES_COPY_BOTH)        = Just CopyBoth
+    fromCInt (#const PGRES_BAD_RESPONSE)     = Just BadResponse
+    fromCInt (#const PGRES_NONFATAL_ERROR)   = Just NonfatalError
+    fromCInt (#const PGRES_FATAL_ERROR)      = Just FatalError
+    fromCInt (#const PGRES_SINGLE_TUPLE)     = Just SingleTuple
+    fromCInt (#const PGRES_PIPELINE_SYNC)    = Just PipelineSync
+    fromCInt (#const PGRES_PIPELINE_ABORTED) = Just PipelineAbort
     fromCInt _ = Nothing
 
 instance ToCInt ExecStatus where
@@ -67,6 +86,8 @@ instance ToCInt ExecStatus where
     toCInt NonfatalError = (#const PGRES_NONFATAL_ERROR)
     toCInt FatalError    = (#const PGRES_FATAL_ERROR)
     toCInt SingleTuple   = (#const PGRES_SINGLE_TUPLE)
+    toCInt PipelineSync  = (#const PGRES_PIPELINE_SYNC)
+    toCInt PipelineAbort = (#const PGRES_PIPELINE_ABORTED)
 
 
 data FieldCode
@@ -230,7 +251,7 @@ instance FromCInt ConnStatus where
     fromCInt (#const CONNECTION_SSL_STARTUP)       = return ConnectionSSLStartup
     -- fromCInt (#const CONNECTION_NEEDED)         = return ConnectionNeeded
     fromCInt _ = Nothing
-    
+
 
 data TransactionStatus
     = TransIdle    -- ^ currently idle
@@ -263,6 +284,25 @@ instance FromCInt Format where
     fromCInt 1 = Just Binary
     fromCInt _ = Nothing
 
+
+-- |
+--
+-- @since 0.11.0.0
+data PipelineStatus
+    = PipelineOn           -- ^ The 'Connection' is in pipeline mode.
+    | PipelineOff          -- ^ The 'Connection' is /not/ in pipeline mode.
+    | PipelineAborted      -- ^ The 'Connection' is in pipeline mode and an error
+                           -- occurred while processing the current pipeline. The
+                           -- aborted flag is cleared when 'getResult' returns a
+                           -- result with status 'PipelineSync'.
+  deriving (Eq, Show)
+
+instance FromCInt PipelineStatus where
+    fromCInt (#const PQ_PIPELINE_ON) = return PipelineOn
+    fromCInt (#const PQ_PIPELINE_OFF) = return PipelineOff
+    fromCInt (#const PQ_PIPELINE_ABORTED) = return PipelineAborted
+    fromCInt _ = Nothing
+
 -------------------------------------------------------------------------------
 -- System.IO enumerations
 -------------------------------------------------------------------------------

src/Database/PostgreSQL/LibPQ/FFI.hs

@@ -302,6 +302,21 @@ foreign import capi        "hs-libpq.h &PQfreemem"
 foreign import capi        "hs-libpq.h PQfreemem"
     c_PQfreemem :: Ptr a -> IO ()
 
+foreign import capi        "hs-libpq.h PQpipelineStatus"
+    c_PQpipelineStatus :: Ptr PGconn -> IO CInt
+
+foreign import capi        "hs-libpq.h PQenterPipelineMode"
+    c_PQenterPipelineMode :: Ptr PGconn -> IO CInt
+
+foreign import capi        "hs-libpq.h PQexitPipelineMode"
+    c_PQexitPipelineMode :: Ptr PGconn -> IO CInt
+
+foreign import capi        "hs-libpq.h PQpipelineSync"
+    c_PQpipelineSync :: Ptr PGconn -> IO CInt
+
+foreign import capi        "hs-libpq.h PQsendFlushRequest"
+    c_PQsendFlushRequest :: Ptr PGconn -> IO CInt
+
 -------------------------------------------------------------------------------
 -- FFI imports: noticebuffers
 -------------------------------------------------------------------------------

test/Smoke.hs

@@ -5,7 +5,6 @@ import Control.Monad             (unless)
 import Data.Foldable             (toList)
 import Database.PostgreSQL.LibPQ
 import System.Environment        (getEnvironment)
-import System.Exit               (exitFailure)
 import Test.Tasty                (defaultMain, testGroup)
 import Test.Tasty.HUnit          (assertEqual, testCaseSteps)
 
@@ -18,6 +17,7 @@ main = do
     withConnstring $ \connString -> defaultMain $ testGroup "postgresql-libpq"
         [ testCaseSteps "smoke" $ smoke connString
         , testCaseSteps "issue54" $ issue54 connString
+        , testCaseSteps "pipeline" $ testPipeline connString
         ]
 
 withConnstring :: (BS8.ByteString -> IO ()) -> IO ()
@@ -57,6 +57,7 @@ smoke connstring info = do
     transactionStatus conn >>= infoShow
     protocolVersion conn   >>= infoShow
     serverVersion conn     >>= infoShow
+    pipelineStatus conn    >>= infoShow
 
     s <- status conn
     assertEqual "connection not ok" ConnectionOk s
@@ -87,3 +88,34 @@ issue54 connString info = do
 
     assertEqual "fst not null" BS.empty val1
     assertEqual "snd not null" BS.empty val2
+
+testPipeline :: BS8.ByteString -> (String -> IO ()) -> IO ()
+testPipeline connstring info = do
+    conn <- connectdb connstring
+
+    setnonblocking conn True `shouldReturn` True
+    enterPipelineMode conn `shouldReturn` True
+    pipelineStatus conn `shouldReturn` PipelineOn
+    sendQueryParams conn (BS8.pack "select 1") [] Text `shouldReturn` True
+    sendQueryParams conn (BS8.pack "select 2") [] Text `shouldReturn` True
+    pipelineSync conn `shouldReturn` True
+
+    Just r1 <- getResult conn
+    resultStatus r1 `shouldReturn` TuplesOk
+    getvalue r1 0 0 `shouldReturn` Just (BS8.pack "1")
+    Nothing <- getResult conn
+
+    Just r2 <- getResult conn
+    getvalue r2 0 0 `shouldReturn` Just (BS8.pack "2")
+    Nothing <- getResult conn
+
+    Just r3 <- getResult conn
+    resultStatus r3 `shouldReturn` PipelineSync
+
+    finish conn
+  where
+    shouldBe r value = assertEqual "shouldBe" r value
+
+    shouldReturn action value = do
+        r <- action
+        r `shouldBe` value