packages feed

pqi-native 1.0.1.15 → 1.0.1.16

raw patch · 4 files changed

+99/−45 lines, 4 filesdep ~pqi-conformancePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: pqi-conformance

API changes (from Hackage documentation)

Files

CHANGELOG.md view
@@ -1,3 +1,11 @@+# v1.0.1.16++## Fixes++- Fixed reading a result's cells being quadratic in its rows ([#15](https://github.com/nikita-volkov/pqi-native/issues/15)). `mkResult` kept the rows, fields and parameter OIDs as lists, and `getvalue`, `getvalue'`, `getisnull` and `getlength` reached the requested row with `drop` from the head of the list on every call, so scanning every row through them took time quadratic in the row count: 3.9 s for 40,000 rows, where libpq, which indexes an array, took 17 ms. The rows, fields and parameter OIDs are now indexed into vectors once per result, on first access, and shared by every accessor, so each lookup takes constant time. Out-of-range indexes still return the same values. Caught by the `pqi-conformance` spec `Pqi.Conformance.Operation.Getvalue.ManyRows`.++- Fixed a synchronous call returning the results of an earlier query whose results were still unread ([#16](https://github.com/nikita-volkov/pqi-native/issues/16)). `exec`, `execParams`, `prepare`, `execPrepared`, `describePrepared` and `describePortal` sent their query and read up to the next `ReadyForQuery` without recording that a query was in flight. When an asynchronous exception, such as one from `timeout`, interrupted one of them while it waited, the interrupted query's messages stayed unread. The next call read them as its own and returned the interrupted query's result, and every call after that returned the result of the one before it, with no error. A `sendQuery` or `sendQueryParams` whose results were never fetched had the same effect on the synchronous call that followed it. libpq's `PQexecParams` is a blocking foreign call that cannot be interrupted, and `PQexecStart` reads and discards any leftover results before a synchronous call sends its query. The synchronous flows now do the same: they first discard any results still pending (outside pipeline mode), and they mark the connection busy from their send until their own `ReadyForQuery`, so an interrupted flow leaves its results to be discarded by the next one. If the connection is lost while discarding, the call returns `Nothing`, as `PQexec` does. Because the connection is now marked busy while a synchronous flow waits, a `cancel` from another thread reaches its query, where before it was skipped as if nothing were running. Caught by the `pqi-conformance` specs `Pqi.Conformance.Operation.ExecParams.Interrupted` and `Pqi.Conformance.Operation.Exec.Interrupted`.+ # v1.0.1.15  ## Fixes
pqi-native.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 1.0.1.15+version: 1.0.1.16 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -160,5 +160,5 @@   build-depends:     base >=4.11 && <5,     hspec >=2.11 && <2.12,-    pqi-conformance ^>=1.0.16,+    pqi-conformance ^>=1.0.17,     pqi-native,
src/library/Pqi/Native/Query.hs view
@@ -17,7 +17,7 @@   ) where -import Control.Exception (catch, mask_)+import Control.Exception (catch, mask, mask_) import qualified Data.Map.Strict as Map import qualified Data.Sequence as Seq import qualified Data.Vector.Mutable as MVector@@ -77,33 +77,41 @@  -- | Simple query. Returns the last result, mirroring @PQexec@. exec :: Connection -> ByteString -> IO (Maybe NativeResult)-exec connection sql = withReady connection $ guardLoss do-  sendMessage connection (queryMessage sql)-  lastMaybe <$> collectSimple connection sql+exec connection sql =+  synchronously+    connection+    (queryMessage sql)+    (lastMaybe <$> collectSimple connection sql)  -- | Parameterized query via the extended protocol. execParams :: Connection -> ByteString -> [Maybe (Word32, ByteString, Format)] -> Format -> IO (Maybe NativeResult) execParams connection sql params resultFormat   | tooManyParams params = pure Nothing-  | otherwise = withReady connection $ guardLoss do-      sendMessage connection (paramsWrite sql params resultFormat)-      Just <$> collectExtended connection sql+  | otherwise =+      synchronously+        connection+        (paramsWrite sql params resultFormat)+        (Just <$> collectExtended connection sql)  -- | Prepare a named statement. prepare :: Connection -> ByteString -> ByteString -> Maybe [Word32] -> IO (Maybe NativeResult) prepare connection name sql parameterTypes   | maybe False tooManyParams parameterTypes = pure Nothing-  | otherwise = withReady connection $ guardLoss do-      sendMessage connection (prepareWrite name sql parameterTypes)-      Just <$> collectExtended connection sql+  | otherwise =+      synchronously+        connection+        (prepareWrite name sql parameterTypes)+        (Just <$> collectExtended connection sql)  -- | Execute a previously prepared statement. execPrepared :: Connection -> ByteString -> [Maybe (ByteString, Format)] -> Format -> IO (Maybe NativeResult) execPrepared connection name params resultFormat   | tooManyParams params = pure Nothing-  | otherwise = withReady connection $ guardLoss do-      sendMessage connection (preparedWrite name params resultFormat)-      Just <$> collectExtended connection ""+  | otherwise =+      synchronously+        connection+        (preparedWrite name params resultFormat)+        (Just <$> collectExtended connection "")  -- * Asynchronous flows @@ -400,15 +408,19 @@  -- | Describe a prepared statement. describePrepared :: Connection -> ByteString -> IO (Maybe NativeResult)-describePrepared connection name = withReady connection $ guardLoss do-  sendMessage connection (describeStatementMessage name <> syncMessage)-  Just <$> collectExtended connection ""+describePrepared connection name =+  synchronously+    connection+    (describeStatementMessage name <> syncMessage)+    (Just <$> collectExtended connection "")  -- | Describe a portal. describePortal :: Connection -> ByteString -> IO (Maybe NativeResult)-describePortal connection name = withReady connection $ guardLoss do-  sendMessage connection (describePortalMessage name <> syncMessage)-  Just <$> collectExtended connection ""+describePortal connection name =+  synchronously+    connection+    (describePortalMessage name <> syncMessage)+    (Just <$> collectExtended connection "")  -- * Parameter projections @@ -435,6 +447,36 @@   Binary -> 1  -- * Result collection++-- | Run a synchronous flow: send the write, then collect its results up to+-- @ReadyForQuery@ (or the start of a @COPY@).+--+-- First, as libpq's @PQexecStart@ does, any results of an earlier query that+-- are still unread are read and discarded, so that the flow collects its own+-- and not theirs. Those are left by a 'sendQuery' whose results were never+-- fetched, and by a synchronous flow that an asynchronous exception+-- interrupted while it waited: the connection is marked busy from the send+-- until the collection is over, so an interrupted flow leaves it busy. This+-- also lets a 'Pqi.cancel' from another thread reach the query. libpq refuses+-- a synchronous flow in pipeline mode, so there is nothing to discard there.+--+-- If the connection is lost while discarding, the flow returns no result and+-- the error message is the loss, as libpq's @PQexec@ does.+synchronously :: Connection -> Poker.Write -> IO (Maybe NativeResult) -> IO (Maybe NativeResult)+synchronously connection write collect = withReady connection do+  pipeline <- inPipeline connection+  unless pipeline discardUnread+  status <- readIORef (connStatus connection)+  if status /= ConnectionOk+    then pure Nothing+    else guardLoss $ mask \restore -> do+      sendMessage connection write+      writeIORef (asyncPending connection) True+      result <- restore collect+      writeIORef (asyncPending connection) False+      pure result+  where+    discardUnread = getNextResult connection >>= maybe (pure ()) (const discardUnread)  -- | Only run a flow on a ready connection; mirror libpq returning no result -- when the connection is not usable, with \"no connection to the server\" as
src/library/Pqi/Native/Types.hs view
@@ -18,6 +18,7 @@ import Data.Char (isDigit) import Data.List (findIndex) import qualified Data.Map.Strict as Map+import qualified Data.Vector as Vector import Pqi   ( Cancel (..),     ExecStatus (..),@@ -171,32 +172,39 @@             (literalErrorMessage result),       resultErrorField = \field -> pure (Map.lookup (fieldCodeByte field) (errorFields result)),       unsafeFreeResult = pure (),-      ntuples = pure (fromIntegral (length (rows result))),-      nfields = pure (fromIntegral (length (fields result))),+      ntuples = pure (fromIntegral (Vector.length indexedRows)),+      nfields = pure (fromIntegral (Vector.length indexedFields)),       fname = \column -> pure $ do-        fd <- atMay (fields result) column+        fd <- atMay indexedFields column         if ByteString.null (name fd) then Nothing else Just (name fd),       fnumber = \colName ->         pure (fromIntegral <$> findIndex (\field -> name field == foldIdentifier colName) (fields result)),-      ftable = \column -> pure (maybe 0 tableOid (atMay (fields result) column)),+      ftable = \column -> pure (maybe 0 tableOid (atMay indexedFields column)),       ftablecol = \column ->-        pure (maybe 0 (\field -> fromIntegral (columnAttributeNumber field :: Int16)) (atMay (fields result) column)),+        pure (maybe 0 (\field -> fromIntegral (columnAttributeNumber field :: Int16)) (atMay indexedFields column)),       fformat = \column ->-        pure (maybe Text (\field -> formatOf (formatCode field)) (atMay (fields result) column)),-      ftype = \column -> pure (maybe 0 typeOid (atMay (fields result) column)),+        pure (maybe Text (\field -> formatOf (formatCode field)) (atMay indexedFields column)),+      ftype = \column -> pure (maybe 0 typeOid (atMay indexedFields column)),       fmod = \column ->-        pure (maybe 0 (\field -> fromIntegral (typeModifier field :: Int32)) (atMay (fields result) column)),+        pure (maybe 0 (\field -> fromIntegral (typeModifier field :: Int32)) (atMay indexedFields column)),       fsize = \column ->-        pure (maybe 0 (\field -> fromIntegral (typeSize field :: Int16)) (atMay (fields result) column)),-      getvalue = \row column -> pure (join (cellAt result row column)),-      getvalue' = \row column -> pure (join (cellAt result row column)),-      getisnull = \row column -> pure (maybe True isNothing (cellAt result row column)),-      getlength = \row column -> pure (maybe 0 (maybe 0 ByteString.length) (cellAt result row column)),-      nparams = pure (fromIntegral (length (paramOids result))),-      paramtype = \index -> pure (fromMaybe 0 (atMay (paramOids result) index)),+        pure (maybe 0 (\field -> fromIntegral (typeSize field :: Int16)) (atMay indexedFields column)),+      getvalue = \row column -> pure (join (cellAt indexedRows row column)),+      getvalue' = \row column -> pure (join (cellAt indexedRows row column)),+      getisnull = \row column -> pure (maybe True isNothing (cellAt indexedRows row column)),+      getlength = \row column -> pure (maybe 0 (maybe 0 ByteString.length) (cellAt indexedRows row column)),+      nparams = pure (fromIntegral (Vector.length indexedParamOids)),+      paramtype = \index -> pure (fromMaybe 0 (atMay indexedParamOids index)),       cmdStatus = pure (Just (fromMaybe "" (commandTag result))),       cmdTuples = pure (Just (maybe "" affectedRows (commandTag result)))     }+  where+    -- Built once per result on first access and shared by every accessor,+    -- so that a lookup by index is constant time, as in libpq. Walking the+    -- lists for each lookup made scanning a result quadratic in its rows.+    indexedRows = Vector.fromList (map Vector.fromList (rows result))+    indexedFields = Vector.fromList (fields result)+    indexedParamOids = Vector.fromList (paramOids result)  -- | Build a 'Cancel' whose field closes over the given 'NativeCancel'. mkCancel :: NativeCancel -> Cancel@@ -219,16 +227,12 @@  -- * Helpers for 'mkResult' -atMay :: [a] -> Int32 -> Maybe a-atMay xs i-  | i < 0 = Nothing-  | otherwise = case drop (fromIntegral i) xs of-      x : _ -> Just x-      [] -> Nothing+atMay :: Vector.Vector a -> Int32 -> Maybe a+atMay xs i = xs Vector.!? fromIntegral i -cellAt :: NativeResult -> Int32 -> Int32 -> Maybe (Maybe ByteString)-cellAt result row column = do-  rowValues <- atMay (rows result) row+cellAt :: Vector.Vector (Vector.Vector (Maybe ByteString)) -> Int32 -> Int32 -> Maybe (Maybe ByteString)+cellAt indexedRows row column = do+  rowValues <- atMay indexedRows row   atMay rowValues column  formatOf :: Int16 -> Format