diff --git a/hasql.cabal b/hasql.cabal --- a/hasql.cabal +++ b/hasql.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql -version: 1.10.3.2 +version: 1.10.3.3 category: Hasql, Database, PostgreSQL synopsis: Fast PostgreSQL driver with a flexible mapping API description: diff --git a/src/benchmarks/Main.hs b/src/benchmarks/Main.hs --- a/src/benchmarks/Main.hs +++ b/src/benchmarks/Main.hs @@ -12,7 +12,10 @@ main :: IO () main = do - Right connection <- acquireConnection + connection <- acquireConnection + connection <- case connection of + Left err -> fail (show err) + Right connection -> pure connection useConnection connection where acquireConnection = @@ -24,7 +27,16 @@ sessionBench "manyLargeResults" sessionWithManyLargeResults, sessionBench "manyLargeResultsViaPipeline" sessionWithManyLargeResultsViaPipeline, sessionBench "manySmallResults" sessionWithManySmallResults, - sessionBench "manySmallResultsViaPipeline" sessionWithManySmallResultsViaPipeline + sessionBench "manySmallResultsViaPipeline" sessionWithManySmallResultsViaPipeline, + bgroup + "singleStatementOverhead" + [ sessionBench "1-sequential" sessionWith1SmallResult, + sessionBench "1-pipeline" sessionWith1SmallResultViaPipeline, + sessionBench "10-sequential" sessionWith10SmallResults, + sessionBench "10-pipeline" sessionWith10SmallResultsViaPipeline, + sessionBench "100-sequential" sessionWithManySmallResults, + sessionBench "100-pipeline" sessionWithManySmallResultsViaPipeline + ] ] where sessionBench :: (NFData a) => String -> B.Session a -> Benchmark @@ -33,33 +45,49 @@ -- * Sessions -sessionWithSingleLargeResultInVector :: B.Session (Vector (Int64, Int64)) +sessionWith1SmallResult :: B.Session (Int32, Int32) +sessionWith1SmallResult = + B.statement () statementWithSingleRow + +sessionWith1SmallResultViaPipeline :: B.Session (Int32, Int32) +sessionWith1SmallResultViaPipeline = + B.pipeline (E.statement () statementWithSingleRow) + +sessionWith10SmallResults :: B.Session [(Int32, Int32)] +sessionWith10SmallResults = + replicateM 10 (B.statement () statementWithSingleRow) + +sessionWith10SmallResultsViaPipeline :: B.Session [(Int32, Int32)] +sessionWith10SmallResultsViaPipeline = + B.pipeline (replicateM 10 (E.statement () statementWithSingleRow)) + +sessionWithSingleLargeResultInVector :: B.Session (Vector (Int32, Int32)) sessionWithSingleLargeResultInVector = B.statement () statementWithManyRowsInVector -sessionWithSingleLargeResultInList :: B.Session [(Int64, Int64)] +sessionWithSingleLargeResultInList :: B.Session [(Int32, Int32)] sessionWithSingleLargeResultInList = B.statement () statementWithManyRowsInList -sessionWithManyLargeResults :: B.Session [Vector (Int64, Int64)] +sessionWithManyLargeResults :: B.Session [Vector (Int32, Int32)] sessionWithManyLargeResults = replicateM 100 (B.statement () statementWithManyRowsInVector) -sessionWithManySmallResults :: B.Session [(Int64, Int64)] +sessionWithManySmallResults :: B.Session [(Int32, Int32)] sessionWithManySmallResults = replicateM 100 (B.statement () statementWithSingleRow) -sessionWithManyLargeResultsViaPipeline :: B.Session [Vector (Int64, Int64)] +sessionWithManyLargeResultsViaPipeline :: B.Session [Vector (Int32, Int32)] sessionWithManyLargeResultsViaPipeline = B.pipeline (replicateM 100 (E.statement () statementWithManyRowsInVector)) -sessionWithManySmallResultsViaPipeline :: B.Session [(Int64, Int64)] +sessionWithManySmallResultsViaPipeline :: B.Session [(Int32, Int32)] sessionWithManySmallResultsViaPipeline = B.pipeline (replicateM 100 (E.statement () statementWithSingleRow)) -- * Statements -statementWithSingleRow :: C.Statement () (Int64, Int64) +statementWithSingleRow :: C.Statement () (Int32, Int32) statementWithSingleRow = C.preparable template encoder decoder where @@ -71,12 +99,12 @@ D.singleRow row where row = - tuple <$> (D.column . D.nonNullable) D.int8 <*> (D.column . D.nonNullable) D.int8 + tuple <$> (D.column . D.nonNullable) D.int4 <*> (D.column . D.nonNullable) D.int4 where tuple !a !b = (a, b) -statementWithManyRows :: (D.Row (Int64, Int64) -> D.Result result) -> C.Statement () result +statementWithManyRows :: (D.Row (Int32, Int32) -> D.Result result) -> C.Statement () result statementWithManyRows decoder = C.preparable template encoder (decoder rowDecoder) where @@ -85,15 +113,15 @@ encoder = conquer rowDecoder = - tuple <$> (D.column . D.nonNullable) D.int8 <*> (D.column . D.nonNullable) D.int8 + tuple <$> (D.column . D.nonNullable) D.int4 <*> (D.column . D.nonNullable) D.int4 where tuple !a !b = (a, b) -statementWithManyRowsInVector :: C.Statement () (Vector (Int64, Int64)) +statementWithManyRowsInVector :: C.Statement () (Vector (Int32, Int32)) statementWithManyRowsInVector = statementWithManyRows D.rowVector -statementWithManyRowsInList :: C.Statement () [(Int64, Int64)] +statementWithManyRowsInList :: C.Statement () [(Int32, Int32)] statementWithManyRowsInList = statementWithManyRows D.rowList diff --git a/src/library-tests/Sharing/ByUnit/Session/StatementSpec.hs b/src/library-tests/Sharing/ByUnit/Session/StatementSpec.hs --- a/src/library-tests/Sharing/ByUnit/Session/StatementSpec.hs +++ b/src/library-tests/Sharing/ByUnit/Session/StatementSpec.hs @@ -16,14 +16,36 @@ describe "Roundtrips" do it "handles simple values correctly" \config -> do Scripts.onPreparableConnection config \connection -> do - let statement = - Statement.preparable - "select $1" - (Encoders.param (Encoders.nonNullable Encoders.int8)) - (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))) - result <- Connection.use connection (Session.statement (42 :: Int64) statement) + result <- Connection.use connection (Session.statement (42 :: Int64) echoStatement) result `shouldBe` Right 42 + it "reuses a prepared statement across executions in one session" \config -> do + -- The first execution is a cache miss (separate PARSE roundtrip), + -- the second a cache hit (single roundtrip). Both must succeed. + Scripts.onPreparableConnection config \connection -> do + result <- + Connection.use connection do + a <- Session.statement (1 :: Int64) echoStatement + b <- Session.statement (2 :: Int64) echoStatement + pure (a, b) + result `shouldBe` Right (1, 2) + + it "keeps a prepared statement usable after an EXECUTE error" \config -> do + -- Regression: PARSE succeeds, EXECUTE fails (division by zero). The + -- statement is on the server under its cached name, so a later use on the + -- same connection must hit the cache rather than re-issuing PARSE for an + -- already-existing name ("prepared statement ... already exists"). + Scripts.onPreparableConnection config \connection -> do + failure <- Connection.use connection (Session.statement 0 divStatement) + failure `shouldSatisfy` isLeft + success <- Connection.use connection (Session.statement 1 divStatement) + success `shouldBe` Right 1 + + it "works on an unpreparable connection" \config -> do + Scripts.onUnpreparableConnection config \connection -> do + result <- Connection.use connection (Session.statement (42 :: Int64) echoStatement) + result `shouldBe` Right 42 + describe "Error Handling" do it "captures query errors correctly" \config -> do Scripts.onPreparableConnection config \connection -> do @@ -40,3 +62,18 @@ case result of Left (Errors.StatementSessionError _ _ _ _ _ (Errors.ServerStatementError _)) -> pure () _ -> expectationFailure $ "Unexpected result: " <> show result + +echoStatement :: Statement.Statement Int64 Int64 +echoStatement = + Statement.preparable + "select $1" + (Encoders.param (Encoders.nonNullable Encoders.int8)) + (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))) + +-- | Parses fine, but fails at execution time when given 0 (division by zero). +divStatement :: Statement.Statement Int64 Int64 +divStatement = + Statement.preparable + "select 1 / $1" + (Encoders.param (Encoders.nonNullable Encoders.int8)) + (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))) diff --git a/src/library/Hasql/Codecs/RequestingOid.hs b/src/library/Hasql/Codecs/RequestingOid.hs --- a/src/library/Hasql/Codecs/RequestingOid.hs +++ b/src/library/Hasql/Codecs/RequestingOid.hs @@ -14,7 +14,6 @@ import Data.HashMap.Strict qualified as HashMap import Hasql.Codecs.RequestingOid.LookingUp qualified as LookingUp import Hasql.Platform.Prelude hiding (lift, lookup) -import PostgreSQL.Binary.Decoding qualified as Binary type RequestingOid = LookingUp.LookingUp diff --git a/src/library/Hasql/Engine/Contexts/Pipeline.hs b/src/library/Hasql/Engine/Contexts/Pipeline.hs --- a/src/library/Hasql/Engine/Contexts/Pipeline.hs +++ b/src/library/Hasql/Engine/Contexts/Pipeline.hs @@ -9,7 +9,6 @@ import Data.HashSet qualified as HashSet import Hasql.Codecs.Encoders.Params qualified as Params import Hasql.Codecs.RequestingOid qualified as RequestingOid -import Hasql.Comms.Recv qualified as Comms.Recv import Hasql.Comms.Roundtrip qualified as Comms.Roundtrip import Hasql.Engine.Decoders.Result qualified as Decoders.Result import Hasql.Engine.Errors qualified as Errors diff --git a/src/library/Hasql/Engine/Contexts/Session.hs b/src/library/Hasql/Engine/Contexts/Session.hs --- a/src/library/Hasql/Engine/Contexts/Session.hs +++ b/src/library/Hasql/Engine/Contexts/Session.hs @@ -1,11 +1,17 @@ module Hasql.Engine.Contexts.Session where +import Data.HashMap.Strict qualified as HashMap +import Data.HashSet qualified as HashSet import Hasql.Codecs.Encoders.Params qualified as Params +import Hasql.Codecs.RequestingOid qualified as RequestingOid import Hasql.Comms.Roundtrip qualified as Comms.Roundtrip import Hasql.Engine.Contexts.Pipeline qualified as Pipeline import Hasql.Engine.Decoders.Result qualified as Decoders.Result import Hasql.Engine.Errors qualified as Errors +import Hasql.Engine.PqProcedures.SelectTypeInfo qualified as PqProcedures.SelectTypeInfo import Hasql.Engine.Structures.ConnectionState qualified as ConnectionState +import Hasql.Engine.Structures.OidCache qualified as OidCache +import Hasql.Engine.Structures.StatementCache qualified as StatementCache import Hasql.Platform.Prelude import Hasql.Pq qualified as Pq @@ -62,7 +68,14 @@ ) -- | --- Execute a statement by providing parameters to it. +-- Execute a single statement by providing parameters to it, +-- running it directly in serial mode. +-- +-- Each execution is a dedicated network roundtrip. The first execution of a +-- preparable statement costs an extra roundtrip (a separate @PARSE@), after +-- which steady-state execution is a single roundtrip. +-- +-- To batch multiple statements into fewer roundtrips, use 'pipeline' instead. statement :: ByteString -> Params.Params params -> @@ -71,7 +84,92 @@ params -> Session result statement sql paramsEncoder decoder preparable params = - pipeline (Pipeline.statement sql paramsEncoder decoder preparable params) + Session \connectionState -> do + let usePreparedStatements = ConnectionState.preparedStatements connectionState + statementCache = ConnectionState.statementCache connectionState + oidCache = ConnectionState.oidCache connectionState + connection = ConnectionState.connection connectionState + requestingDecoder = Decoders.Result.unwrap decoder + unknownTypes = + Params.toUnknownTypes paramsEncoder + <> RequestingOid.toUnknownTypes requestingDecoder + missingTypes = OidCache.selectUnknownNames unknownTypes oidCache + oidCacheUpdates <- + PqProcedures.SelectTypeInfo.run connection (PqProcedures.SelectTypeInfo.SelectTypeInfo missingTypes) + case oidCacheUpdates of + Left err -> pure (Left err, connectionState) + Right oidCacheUpdates -> do + -- Validate that all requested types were found. + let foundTypes = HashMap.keysSet oidCacheUpdates + notFoundTypes = HashSet.difference missingTypes foundTypes + if not (HashSet.null notFoundTypes) + then pure (Left (Errors.MissingTypesSessionError notFoundTypes), connectionState) + else do + let newOidCache = oidCache <> OidCache.fromHashMap oidCacheUpdates + oidHashMap = OidCache.toHashMap newOidCache + decoder' = RequestingOid.toBase requestingDecoder oidHashMap + prepared = usePreparedStatements && preparable + -- Single-statement context for error reporting: + -- total statements 1, index 0. + context = Just (1, 0, sql, Params.renderReadable paramsEncoder params, prepared) + mapError = \case + Comms.Roundtrip.ClientError _ details -> + Errors.ConnectionSessionError (maybe "" decodeUtf8Lenient details) + Comms.Roundtrip.ServerError recvError -> + Errors.fromRecvError recvError + withState (result, newStatementCache) = + ( first mapError result, + connectionState + { ConnectionState.oidCache = newOidCache, + ConnectionState.statementCache = newStatementCache + } + ) + fmap withState + $ if prepared + then do + let (oidList, valueAndFormatList) = + Params.compilePreparedStatementData paramsEncoder oidHashMap params + pqOidList = fmap (Pq.Oid . fromIntegral) oidList + encodedParams = + valueAndFormatList + & fmap (fmap (\(bytes, format) -> (bytes, bool Pq.Binary Pq.Text format))) + execute remoteKey = + Comms.Roundtrip.toSerialIO + (Comms.Roundtrip.queryPrepared context remoteKey encodedParams Pq.Binary decoder') + connection + case StatementCache.lookup sql pqOidList statementCache of + Just remoteKey -> do + result <- execute remoteKey + pure (result, statementCache) + Nothing -> do + let (remoteKey, newStatementCache) = StatementCache.insert sql pqOidList statementCache + -- In non-pipeline mode PARSE and EXECUTE cannot be sent + -- back-to-back, so prepare in a dedicated roundtrip first. + prepareResult <- + Comms.Roundtrip.toSerialIO + (Comms.Roundtrip.prepare context remoteKey sql pqOidList) + connection + case prepareResult of + -- PARSE failed: the statement is not on the server, so + -- keep the old cache (no entry committed). + Left err -> pure (Left err, statementCache) + Right () -> do + -- PARSE succeeded, so the statement is on the server + -- under remoteKey regardless of whether EXECUTE then + -- fails. Commit the cache so a later use hits it instead + -- of re-issuing PARSE for an already-existing name. + result <- execute remoteKey + pure (result, newStatementCache) + else do + let encodedParams = + params + & Params.compileUnpreparedStatementData paramsEncoder oidHashMap + & fmap (fmap (\(oid, bytes, format) -> (Pq.Oid (fromIntegral oid), bytes, bool Pq.Binary Pq.Text format))) + result <- + Comms.Roundtrip.toSerialIO + (Comms.Roundtrip.queryParams context sql encodedParams Pq.Binary decoder') + connection + pure (result, statementCache) -- | -- Execute a pipeline.