pgstream (empty) → 0.1.0.3
raw patch · 12 files changed
+1527/−0 lines, 12 filesdep +asyncdep +attoparsecdep +basesetup-changed
Dependencies added: async, attoparsec, base, blaze-builder, bytestring, conduit, conduit-extra, deepseq, mtl, parallel, postgresql-binary, postgresql-libpq, resource-pool, resourcet, scientific, stm, stm-chans, stm-conduit, stringsearch, template-haskell, text, time, transformers, uuid, vector
Files
- LICENSE +2/−0
- README.md +195/−0
- Setup.hs +2/−0
- cbits/array_conversion.c +149/−0
- cbits/array_conversion.h +9/−0
- pgstream.cabal +100/−0
- src/Database/PostgreSQL/Stream.hs +281/−0
- src/Database/PostgreSQL/Stream/Connection.hs +71/−0
- src/Database/PostgreSQL/Stream/FromRow.hs +409/−0
- src/Database/PostgreSQL/Stream/Parallel.hs +34/−0
- src/Database/PostgreSQL/Stream/QueryBuilder.hs +187/−0
- src/Database/PostgreSQL/Stream/Types.hs +88/−0
+ LICENSE view
@@ -0,0 +1,2 @@+Copyright (c) 2015 Elsen Inc.+All rights reserved.
+ README.md view
@@ -0,0 +1,195 @@+<p align="center">+ <a href="http://elsen.co">+ <img src="https://elsen.co/img/apple-touch-icon-144x144.png"/>+ </a>+</p>++pgstream+========++Streaming of Postgres through the binary protocol into Haskell. Uses attoparsec+and some hand-written kernels for array extraction. Results are streamed into+vectors or batched into serial or parallel+[Conduit](https://hackage.haskell.org/package/conduit-1.2.4/docs/Data-Conduit.html)+pipelines for stream composition.++[](https://travis-ci.org/elsen-trading/pgstream)++Installation+------------++```bash+$ cabal install pg_stream.cabal+```++Usage+-----++**Connections**++Connections to Postgres are established with the ``connect`` function yielding+the connection object.++```haskell+connect :: ConnSettings -> IO PQ.Connection+```++Connections are specified by a ConnSettings.++```haskell+creds :: ConnSettings+creds = ConnSettings {_host = "localhost", _dbname="invest", _user="dbadmin"}+```++Connections are pooled per process. Connection pooling is specified by three+parameters.++* **Stripes**: Stripe count. The number of distinct sub-pools to maintain. The smallest acceptable value is 1.+* **Keep Alive**: Amount of time for which an unused resource is kept open. The smallest acceptable value is 0.5 seconds.+* **Affinity**: Maximum number of resources to keep open per stripe. The smallest acceptable value is 1.++The default settings are:++```haskell+defaultPoolSettings :: PoolSettings+defaultPoolSettings = PoolSettings { _stripes = 1, _keepalive = 10, _affinity = 10 }+```+++**Queries**++Queries are executed using ``query`` for statements that yield result sets or by+``execute`` for queries that return a status code.++```haskell+query :: (FromRow r, ToSQL a) => PQ.Connection -> Query -> a -> IO [r]+execute :: (ToSQL a) => PQ.Connection -> Query -> a -> IO ()+```++For example:++```haskell+run :: IO [Row]+run = do+ conn <- connect creds+ query conn sample args+```++SQL queries are constructed via quasiquoter (``[sql| ... |]``) which generates a+``Query`` (newtype around a bytestring). Values and SQL fragments can be spliced+into this template as arguments.++```haskell+{-# LANGUAGE QuasiQuotes #-}++sample :: Query+sample = [sql|+ SELECT+ deltas.sid AS sid,+ EXTRACT(EPOCH FROM deltas.day) AS day,+ (ohlcs :: float4[])+ FROM deltas+ INNER JOIN security_groupings ON deltas.sid = security_groupings.sid+ INNER JOIN currentprice ON (+ deltas.sid = currentprice.sid+ AND deltas.DAY = currentprice.DAY+ AND currentprice.val BETWEEN 0 AND 500+ )+ WHERE security_groupings.name = 'SP900'+ AND deltas.day BETWEEN TO_TIMESTAMP({1}) AND TO_TIMESTAMP({2})+ ORDER BY deltas.sid,+ deltas.DAY ASC+ {3}+ ;+|]+```++**Arguments**++If the types of arguments are constrained by inference then no annotations are+necessary. Otherwise annotations are needed to refine the Num/String instances+into concrete types so they can be serialized and sent to Postgres.++```haskell+args :: (Int, Int, SQL)+args = ( 1335855600 , 1336374000 , "LIMIT 100000")+```++The conversion from Haskell to Postgres types is defined by the+FromField/ToField typeclasses with the mapping given by.++| Postgres | Haskell |+| ------------- |---------------|+| int2 | Int8 |+| int4 | Int32 |+| int8 | Int64 |+| float4 | Float |+| float8 | Double |+| numeric | Scientific |+| uuid | UUID |+| char | Char |+| text | Text |+| date | Day |+| bytea | ByteString |+| bool | Bool |+| int4[] | Vector Int32 |+| float4[] | Vector Float |+| money | Fixed E3 |+| null a | Maybe a |++If the result set type is given as ``Maybe a`` then any missing value are+manifest as ``Nothing`` values. And all concrete values are ``Just``.+Effectively makes errors from null values used in unchecked logic+unrepresentable as any function which consumes a potentially nullable field is+forced by the type system to handle both cases.++**Streaming**++```haskell+stream :: (FromRow r, ToSQL a, MonadBaseControl IO m, MonadIO m) =>+ PQ.Connection -- ^ Connection+ -> Query -- ^ Query+ -> a -- ^ Query arguments+ -> Int -- ^ Batch size+ -> C.Source m [r] -- ^ Source conduit+```++Parallel streams can be composed together Software Transactional Memory (STM)+threads to synchronize the polling.++```haskell+import Database.PostgreSQL.Stream.Parallel++parallelStream ::+ PQ.Connection+ -> (PQ.Connection -> Source (ResourceT IO) a) -- Source+ -> Sink a (ResourceT IO) () -- Sink+ -> IO ()+```++Development+-----------++```bash+$ cabal sandbox init+$ cabal install --only-dependencies+$ cabal build+```++To attach to the Elsen compute engine:++```bash+$ cabal sandbox add-source path_to_tree+```++Documentation+--------------++```bash+$ cabal haddock+```++Legal+-----++Copyright (c) 2015 Elsen Inc. All rights reserved.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cbits/array_conversion.c view
@@ -0,0 +1,149 @@+#include <stdlib.h>+#include <stdio.h>+#include <stdint.h>+#include <math.h>+#include <arpa/inet.h>+#include <postgresql/libpq-fe.h>++#define INT4OID 23+#define FLOAT4OID 700+#define FLOAT8OID 701+#define NUMERICOID 1700++// XXX: Check endianness problems on OS X? Production is invariable Linux so+// probably shouldn't matter.++// int4[]+struct array_int4 {+ int32_t ndim;+ int32_t _ign;+ Oid elemtype;+ + int32_t size;+ int32_t index;+ int32_t data;+};++// float4[]+struct array_float {+ int32_t ndim;+ int32_t _ign;+ Oid elemtype;+ + int32_t size;+ int32_t index;+ float data;+};++// float4[]+struct array_double {+ int32_t ndim;+ int32_t _ign;+ Oid elemtype;+ + int32_t size;+ int32_t index;+ double data;+};+++// Postgres binary uses network byte order, we have to swap the bytes for the+// floating point values.+static inline float swap( const float inFloat )+{+ float retVal;+ char *in = ( char * ) & inFloat;+ char *out = ( char * ) & retVal;+ + out[0] = in[3];+ out[1] = in[2];+ out[2] = in[1];+ out[3] = in[0];+ + return retVal;+}++#define INT_SEP -1++int extract_int4_array (char *raw_array, int32_t *values, int size)+{+ struct array_int4 *array = (struct array_int4 *) raw_array;+ int32_t *val = &(array->data);+ int32_t ival;+ + if (ntohl(array->ndim) != 1 || ntohl(array->elemtype) != INT4OID) {+ return -1;+ }+ int array_elements = ntohl(array->size);+ + int n = 0;+ val = &(array->data);+ for (int i = 0; i < array_elements; ++i) {+ ival = ntohl (*val);+ if (ival != INT_SEP) {+ ++val;+ values[n] = ntohl(*val);+ n += 1;+ }+ + ++val;+ }+ + return size;+}++#define FLOAT_SEP 6.0e-45++int extract_float_array(char *raw_array, float *values, int size)+{+ struct array_float *array = (struct array_float *) raw_array;+ float *val = &(array->data);+ int32_t ival;+ + if (ntohl(array->ndim) != 1 || ntohl(array->elemtype) != FLOAT4OID) {+ return -1;+ }+ int array_elements = ntohl(array->size);+ + int n = 0;+ val = &(array->data);+ for (int i = 0; i < array_elements; ++i) {+ ival = swap(*val);+ if (ival != FLOAT_SEP) {+ ++val;+ values[n] = swap(*val);+ n += 1;+ }+ ++val;+ }+ + return size;+}++#define DOUBLE_SEP 6.0e-45++int extract_double_array(char *raw_array, double *values, int size)+{+ struct array_double *array = (struct array_double *) raw_array;+ double *val = &(array->data);+ int32_t ival;+ + if (ntohl(array->ndim) != 1 || ntohl(array->elemtype) != FLOAT8OID) {+ return -1;+ }+ int array_elements = ntohl(array->size);+ + int n = 0;+ val = &(array->data);+ for (int i = 0; i < array_elements; ++i) {+ ival = swap(*val);+ if (ival != DOUBLE_SEP) {+ ++val;+ values[n] = swap(*val);+ n += 1;+ }+ ++val;+ }+ + return size;+}
+ cbits/array_conversion.h view
@@ -0,0 +1,9 @@+#include <stdlib.h>+#include <stdio.h>+#include <stdint.h>+#include <arpa/inet.h>+#include <postgresql/libpq-fe.h>++int extract_int4_array (char *raw_array, int32_t *values, int size);+int extract_float_array (char *raw_array, float *values, int size);+int extract_double_array (char *raw_array, double *values, int size);
+ pgstream.cabal view
@@ -0,0 +1,100 @@+name: pgstream+version: 0.1.0.3+license: BSD3+copyright: Copyright 2015 Elsen Inc.+category: Finance+description: Elsen Accelerated Computing Engine+synopsis: Elsen Accelerated Computing Engine+license-file: LICENSE+author: Elsen Inc.+maintainer: info@elsen.co+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10+extra-source-files: + cbits/array_conversion.h++Source-Repository head+ Type: git+ Location: git@github.com:elsen-trading/pgstream.git++-- Compiler flags+flag optimized+ manual: True+ default: True+ description: Compile with optimizations enabled.++flag j+ manual: True+ default: False+ description: Compile in parallel.++-- Executable+library + exposed-modules:+ Database.PostgreSQL.Stream+ Database.PostgreSQL.Stream.Parallel+ Database.PostgreSQL.Stream.Connection+ Database.PostgreSQL.Stream.Types+ Database.PostgreSQL.Stream.FromRow+ Database.PostgreSQL.Stream.QueryBuilder++ build-depends: + base >= 4.6 && <4.10,++ -- Standard library+ text >= 1.1 && <1.2,+ bytestring >= 0.10 && <0.11,+ attoparsec >= 0.12 && <0.13,+ vector >= 0.10 && <0.11,+ uuid >= 1.3 && <1.4,+ scientific >= 0.3 && <0.4, + stringsearch >= 0.3 && <0.4,+ time >= 1.4 && <1.6,+ deepseq >= 1.3 && <1.5,++ -- Streaming+ conduit >= 1.2.4 && <1.3,+ conduit-extra >= 1.1 && <1.2,+ resource-pool >= 0.2 && <0.3,+ blaze-builder >= 0.4 && <0.5, + resourcet >= 1.1 && <1.2,++ -- SQL+ postgresql-libpq >= 0.9 && <0.10,+ postgresql-binary >= 0.5 && <0.6,++ -- Parallelism+ async >= 2.0 && <2.1,+ stm >= 2.2 && <2.5,+ parallel >= 3.2 && <3.3,+ stm-conduit >= 2.5 && <2.6,+ stm-chans >= 2.0 && <3.1,++ -- Text+ template-haskell >= 2.8 && <3.0,++ -- Control flow+ mtl >= 2.2 && <3.0,+ transformers >= 0.4 && <0.5++ -- Testing+ -- criterion -any++ if flag(j) && impl(ghc >= 7.8)+ ghc-options: -j3++ C-sources:+ cbits/array_conversion.c++ cc-options: -std=c99 -O2 -fPIC+ Include-dirs: cbits+ ghc-options:+ -O2+ -funbox-strict-fields+ "-with-rtsopts=-N"+ -rtsopts+ -threaded ++ default-language: Haskell2010+ hs-source-dirs: src
+ src/Database/PostgreSQL/Stream.hs view
@@ -0,0 +1,281 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Database.PostgreSQL.Stream (+ -- ** SQL query ( result set )+ query,+ query_,++ -- ** SQL query ( no result )+ execute,+ execute_,++ -- ** SQL query ( cursor )+ stream,+ stream_,++ -- ** Transacations+ commit,+ rollback,++ -- ** Database pool+ PoolSettings(..),+ pgPool,+ withPgConnection,++ -- ** LibPQ Rexports+ PQ.Connection,+ PQ.ExecStatus(..),+ PQ.ConnStatus(..),++ -- ** Connection initialization+ ConnSettings(..),+ connect,+ connect_alt,++ -- ** SQL quasiquoter+ sql,++ -- ** SQL formatting+ fmtSQL,+ fmtQuery,+ printSQL,++ --ver,+ module Database.PostgreSQL.Stream.Types+) where++import Database.PostgreSQL.Stream.Types+import Database.PostgreSQL.Stream.FromRow+import Database.PostgreSQL.Stream.QueryBuilder+import Database.PostgreSQL.Stream.Connection++import qualified PostgreSQLBinary.Decoder as PD+import qualified Database.PostgreSQL.LibPQ as PQ++import Data.Int+import Data.Text+import Data.Pool+import Data.Monoid+import Data.ByteString (ByteString)+import Data.ByteString.Char8 as B8+import qualified Data.Vector.Storable as V+import Data.UUID.V4(nextRandom)+import Data.UUID(toASCIIBytes, toWords)++import Control.Exception as E+import Control.Monad.Trans+import Control.Applicative+import Control.Monad.Trans.Resource (ResourceT, MonadBaseControl)++import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL++import Data.Version (showVersion)+import qualified Paths_pgstream as Paths++ver :: String+ver = showVersion Paths.version++-------------------------------------------------------------------------------+-- Query+-------------------------------------------------------------------------------++-- Execute SQL query with arguments.+query :: (FromRow r, ToSQL q) => PQ.Connection -> Query -> q -> IO [r]+query conn q args = do+ -- Formatted query+ let query = fmtQuery q args+ -- Execute the query+ result <- PQ.execParams conn query [] PQ.Binary+ case result of+ Nothing -> throw (QueryError "Query execution error." q)+ Just pqres -> do+ status <- PQ.resultStatus pqres+ onError pqres q+ parseRows q pqres++-- Execute SQL query without arguments.+query_ :: (FromRow r) => PQ.Connection -> Query -> IO [r]+query_ conn q = do+ -- Execute the query+ result <- PQ.execParams conn (fromQuery q) [] PQ.Binary+ case result of+ Nothing -> throw (QueryError "Query execution error." q)+ Just pqres -> do+ status <- PQ.resultStatus pqres+ onError pqres q+ parseRows q pqres++-------------------------------------------------------------------------------+-- Execute+-------------------------------------------------------------------------------++execute :: (ToSQL q) => PQ.Connection -> Query -> q -> IO PQ.ExecStatus+execute conn q args = do+ -- Formatted query+ let query = fmtQuery q args+ -- Execute the query+ result <- PQ.execParams conn query [] PQ.Binary+ case result of+ Nothing -> throw (QueryError "Query execution error." q)+ Just pqres -> do+ status <- PQ.resultStatus pqres+ onError pqres q+ return status++execute_ :: PQ.Connection -> Query -> IO PQ.ExecStatus+execute_ conn q = do+ -- Execute the query+ result <- PQ.execParams conn (fromQuery q) [] PQ.Binary+ case result of+ Nothing -> throw (QueryError "Query execution error." q)+ Just pqres -> do+ status <- PQ.resultStatus pqres+ onError pqres q+ return status++-------------------------------------------------------------------------------+-- Error Handling+-------------------------------------------------------------------------------++onError :: PQ.Result -> Query -> IO ()+onError pqres q = do+ status <- PQ.resultStatus pqres+ case status of+ PQ.FatalError -> do+ res <- PQ.resultErrorMessage pqres+ case res of+ Nothing -> return () -- should never happen+ Just err -> throw (QueryError (B8.unpack err) q)+ _ -> return ()++-------------------------------------------------------------------------------+-- Transaction+-------------------------------------------------------------------------------++data IsolationLevel+ = DefaultIsolationLevel+ | ReadCommitted+ | RepeatableRead+ | Serializable+ deriving (Show, Eq, Ord, Enum, Bounded)++data ReadWriteMode+ = DefaultReadWriteMode+ | ReadWrite+ | ReadOnly+ deriving (Show, Eq, Ord, Enum, Bounded)++data TransactionMode = TransactionMode+ { isolationLevel :: !IsolationLevel+ , readWriteMode :: !ReadWriteMode+ } deriving (Show, Eq)++beginMode :: TransactionMode -> PQ.Connection -> IO PQ.ExecStatus+beginMode mode conn = do+ let begin_query = (Query (mconcat ["BEGIN", isolevel, readmode, ";"]))+ execute_ conn begin_query++ where+ isolevel =+ case isolationLevel mode of+ DefaultIsolationLevel -> ""+ ReadCommitted -> " ISOLATION LEVEL READ COMMITTED"+ RepeatableRead -> " ISOLATION LEVEL REPEATABLE READ"+ Serializable -> " ISOLATION LEVEL SERIALIZABLE"+ readmode =+ case readWriteMode mode of+ DefaultReadWriteMode -> ""+ ReadWrite -> " READ WRITE"+ ReadOnly -> " READ ONLY"++defaultTransactionMode :: TransactionMode+defaultTransactionMode = TransactionMode+ defaultIsolationLevel+ defaultReadWriteMode++defaultIsolationLevel ::IsolationLevel+defaultIsolationLevel = DefaultIsolationLevel++defaultReadWriteMode :: ReadWriteMode+defaultReadWriteMode = DefaultReadWriteMode++-- | Rollback a transaction.+rollback :: PQ.Connection -> IO ()+rollback conn = execute_ conn "ABORT" >> return ()++-- | Commit a transaction.+commit :: PQ.Connection -> IO ()+commit conn = execute_ conn "COMMIT" >> return ()++-- | Begin a transaction.+begin :: PQ.Connection -> IO PQ.ExecStatus+begin = beginMode defaultTransactionMode++withTransactionMode :: TransactionMode -> PQ.Connection -> IO a -> IO a+withTransactionMode mode conn act =+ mask $ \restore -> do+ beginMode mode conn+ r <- restore act `E.onException` rollback conn+ commit conn+ return r++-------------------------------------------------------------------------------+-- Streaming Queries+-------------------------------------------------------------------------------++newCursor :: IO (Identifier)+newCursor = do+ uid <- nextRandom+ let (a,b,c,d) = toWords uid+ let bshow = B8.pack . show+ return $ Identifier ("cursor" <> bshow a <> bshow b <> bshow c <> bshow d)++stream :: (FromRow r, ToSQL a, MonadBaseControl IO m, MonadIO m) =>+ PQ.Connection -- ^ Connection+ -> Query -- ^ Query+ -> a -- ^ Query arguments+ -> Int -- ^ Batch size+ -> C.Source m [r] -- ^ Source conduit+stream conn q args n = do+ -- Generate a new cursor from a uuid+ cursor_name <- liftIO $ newCursor+ liftIO $ beginMode defaultTransactionMode conn++ let subsql = fmtQuery q args+ let cursor_query = Query (mconcat [ "DECLARE {1} NO SCROLL CURSOR FOR ", subsql ])++ -- Execute the cursor setup+ liftIO $ execute conn cursor_query (Only cursor_name)++ let fetch_cursor = [sql|FETCH FORWARD {1} FROM {2};|]++ res <- C.tryC (loop conn fetch_cursor cursor_name n)+ case res of+ Left (err :: QueryError) -> liftIO $ do+ -- On SQL exception, rollback the transaction abort the conduit+ rollback conn+ throwIO err+ -- On SQL success, commit+ Right _ -> liftIO $ commit conn++ where+ loop conn q cursor_name n = do+ res <- liftIO $ query conn q (n, cursor_name)+ case res of+ [] -> return ()+ _ -> C.yield res >> loop conn q cursor_name n++stream_ :: (FromRow r, MonadBaseControl IO m, MonadIO m) =>+ PQ.Connection -- ^ Connection+ -> Query -- ^ Query+ -> Int -- ^ Batch size+ -> C.Source m [r] -- ^ Source conduit+stream_ conn q n = stream conn q () n++printSQL :: Query -> IO ()+printSQL (Query bs) = B8.putStrLn bs
+ src/Database/PostgreSQL/Stream/Connection.hs view
@@ -0,0 +1,71 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}++module Database.PostgreSQL.Stream.Connection (+ PoolSettings(..),+ ConnSettings(..),+ defaultPoolSettings,++ pgPool,+ pgPoolSettings,+ withPgConnection,++ connect,+ connect_alt,+) where++import Data.Monoid+import Data.Pool+import Control.Applicative+import Data.Time.Clock (NominalDiffTime)++import Data.ByteString (ByteString)+import qualified Database.PostgreSQL.LibPQ as PQ++-------------------------------------------------------------------------------+-- Connection Pools+-------------------------------------------------------------------------------++data PoolSettings = PoolSettings+ { _stripes :: Int -- ^ Stripe count. The number of distinct sub-pools to maintain. The smallest acceptable value is 1.+ , _keepalive :: NominalDiffTime -- ^ Amount of time for which an unused resource is kept open. The smallest acceptable value is 0.5 seconds.+ , _affinity :: Int -- ^ Maximum number of resources to keep open per stripe. The smallest acceptable value is 1.+ } deriving (Eq, Ord, Show)++defaultPoolSettings :: PoolSettings+defaultPoolSettings = PoolSettings { _stripes = 1, _keepalive = 10, _affinity = 10 }++pgPool :: PQ.Connection -> IO (Pool PQ.Connection)+pgPool conn = createPool (pure conn) PQ.finish 1 10 10++pgPoolSettings :: PoolSettings -> PQ.Connection -> IO (Pool PQ.Connection)+pgPoolSettings PoolSettings{..} conn = createPool (pure conn) PQ.finish _stripes _keepalive _affinity++withPgConnection :: PQ.Connection -> (PQ.Connection -> IO b) -> IO b+withPgConnection conn action = do+ pool <- pgPool conn+ withResource pool action++data ConnSettings = ConnSettings+ { _host :: ByteString+ , _dbname :: ByteString+ , _user :: ByteString+ , _password :: Maybe ByteString+ } deriving (Eq, Ord, Show, Read)++_connect :: ByteString -> IO (Either PQ.ConnStatus PQ.Connection)+_connect connstr = do+ conn <- PQ.connectdb connstr+ rc <- PQ.status conn+ case rc of+ PQ.ConnectionOk -> return (Right conn)+ _ -> return (Left rc)++connect_alt :: ByteString -> IO (Either PQ.ConnStatus PQ.Connection)+connect_alt = _connect++connect :: ConnSettings -> IO (Either PQ.ConnStatus PQ.Connection)+connect (ConnSettings host db user Nothing) = _connect $+ mconcat [ "dbname=" <> db , " host=" <> host , " user=" <> user ]+connect (ConnSettings host db user (Just password)) =_connect $+ mconcat [ "dbname=" <> db , " host=" <> host , " user=" <> user, " password=" <> password ]
+ src/Database/PostgreSQL/Stream/FromRow.hs view
@@ -0,0 +1,409 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Database.PostgreSQL.Stream.FromRow (+ FromRow(..),+ FromField(..),+ Row(..),+ field,++ runRowParser,+ parseRows,+) where++import Database.PostgreSQL.Stream.Types++import Data.Int++import Control.Applicative+import Control.Exception+import Control.Monad (when)+import Control.Monad.Trans+import Control.Monad.Trans.State+import Control.Monad.Trans.Reader++import Data.UUID (UUID)+import Data.Word (Word8)+import Data.Time.Clock+import Data.Time.LocalTime+import Data.Fixed+import Data.Time.Calendar+import Data.Scientific (Scientific)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as BL+import Data.ByteString.Internal (toForeignPtr)+import qualified Data.Vector.Storable as V+import qualified Data.Vector.Storable.Mutable as VM++import Data.Text (Text)+import qualified Data.Text.Lazy as TL++import qualified PostgreSQLBinary.Decoder as PD+import qualified Database.PostgreSQL.LibPQ as PQ++import Unsafe.Coerce+import System.IO.Unsafe++import Foreign.C+import Foreign.Ptr+import Foreign.ForeignPtr.Safe++-------------------------------------------------------------------------------+-- FFI+-------------------------------------------------------------------------------++foreign import ccall unsafe "array_conversion.h extract_int4_array"+ extract_int4_array :: Ptr Word8 -> Ptr CInt -> CInt -> IO Int+foreign import ccall unsafe "array_conversion.h extract_float_array"+ extract_numeric_array :: Ptr Word8 -> Ptr CInt -> CInt -> IO Int++type ExFun = Ptr Word8 -> Ptr CInt -> CInt -> IO Int++-------------------------------------------------------------------------------+-- Rows+-------------------------------------------------------------------------------++class FromRow a where+ fromRow :: RowParser a++instance (FromField a) => FromRow (Only a) where+ fromRow = Only <$> field++instance (FromField a, FromField b) => FromRow (a,b) where+ fromRow = (,) <$> field <*> field++instance (FromField a, FromField b, FromField c) => FromRow (a,b,c) where+ fromRow = (,,) <$> field <*> field <*> field++instance (FromField a, FromField b, FromField c, FromField d) => FromRow (a,b,c,d) where+ fromRow = (,,,) <$> field <*> field <*> field <*> field++instance (FromField a, FromField b, FromField c, FromField d, FromField e) => FromRow (a,b,c,d,e) where+ fromRow = (,,,,) <$> field <*> field <*> field <*> field <*> field++instance (FromField a, FromField b, FromField c, FromField d, FromField e, FromField f) => FromRow (a,b,c,d,e,f) where+ fromRow = (,,,,,) <$> field <*> field <*> field <*> field <*> field <*> field++instance (FromField a, FromField b, FromField c, FromField d, FromField e, FromField f, FromField g) => FromRow (a,b,c,d,e,f,g) where+ fromRow = (,,,,,,) <$> field <*> field <*> field <*> field <*> field <*> field <*> field++-------------------------------------------------------------------------------+-- FieldConversion+-------------------------------------------------------------------------------++data Row = Row+ { row :: {-# UNPACK #-} !PQ.Row+ , rowresult :: !PQ.Result+ }++type FieldParser a = (PQ.Oid, Int, Maybe ByteString) -> a++class FromField a where+ -- Unchecked type conversion from raw postgres bytestring+ fromField :: (PQ.Oid, Int, Maybe ByteString) -> a++-- int2+instance FromField Int8 where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null int2"++-- int4+instance FromField Int16 where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null int4"++-- int4+instance FromField Int32 where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null int4"++-- int8+instance FromField Int64 where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null int8"++-- int8+instance FromField Int where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null int4"++-- float4+instance FromField Float where+ fromField (ty, length, Just bs) = case PD.float4 bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null float4"++-- float8+instance FromField Double where+ fromField (ty, length, Just bs) = case PD.float8 bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null float8"++-- integer+instance FromField Integer where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null integer"++-- numeric+instance FromField Scientific where+ fromField (ty, length, Just bs) = case PD.numeric bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null numeric"++-- uuid+instance FromField UUID where+ fromField (ty, length, Just bs) = case PD.uuid bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null uuid"++-- char+instance FromField Char where+ fromField (ty, length, Just bs) = case PD.char bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null char"++-- text+instance FromField Text where+ fromField (ty, length, Just bs) = case PD.text bs of+ Left x -> throw $ ConversionError "Malformed bytestring."+ Right x -> x+ fromField _ = throw $ ConversionError "Excepted non-null text"++instance FromField TL.Text where+ fromField (ty, length, Just bs) = case PD.text bs of+ Left x -> throw $ ConversionError "Malformed bytestring."+ Right x -> TL.fromStrict x+ fromField _ = throw $ ConversionError "Excepted non-null text"++-- bytea+instance FromField ByteString where+ fromField (ty, length, Just bs) = case PD.bytea bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null bytea"++instance FromField BL.ByteString where+ fromField (ty, length, Just bs) = case PD.bytea bs of { Right x -> BL.fromStrict x }+ fromField _ = throw $ ConversionError "Excepted non-null bytea"++-- bool+instance FromField Bool where+ fromField (ty, length, Just bs) = case PD.bool bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null bool"++-- date+instance FromField Day where+ fromField (ty, length, Just bs) = case PD.date bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++-- time+instance FromField TimeOfDay where+ fromField (ty, length, Just bs) = case PD.time True bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++instance FromField (TimeOfDay, TimeZone) where+ fromField (ty, length, Just bs) = case PD.timetz True bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++instance FromField UTCTime where+ fromField (ty, length, Just bs) = case PD.timestamptz True bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++instance FromField NominalDiffTime where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> (fromIntegral (x :: Int)) }+ fromField _ = throw $ ConversionError "Excepted non-null date"++instance FromField DiffTime where+ fromField (ty, length, Just bs) = case PD.interval True bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++instance FromField LocalTime where+ fromField (ty, length, Just bs) = case PD.timestamp True bs of { Right x -> x }+ fromField _ = throw $ ConversionError "Excepted non-null date"++-- money+instance FromField (Fixed E3) where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> fromIntegral (x :: Int) / 100 }+ fromField _ = throw $ ConversionError "Excepted non-null money"++instance FromField (Fixed E2) where+ fromField (ty, length, Just bs) = case PD.int bs of { Right x -> fromIntegral (x :: Int) / 100 }+ fromField _ = throw $ ConversionError "Excepted non-null money"++-- nullable+instance FromField a => FromField (Maybe a) where+ fromField (_, _, Nothing) = Nothing+ fromField x = Just (fromField x)++-- int4[]+instance FromField (V.Vector Int32) where+ fromField (ty, arrlength, Just bs) = unsafeDupablePerformIO $ int4vector bs arrlength+ fromField _ = throw $ ConversionError "Excepted non-null int4[]"++-- float4[]+instance FromField (V.Vector Float) where+ fromField (ty, arrlength, Just bs) = unsafeDupablePerformIO $ float32vector bs arrlength+ fromField _ = throw $ ConversionError "Excepted non-null float4[]"++-- Converter that is independent of the given length and type+converter :: Show e => (ByteString -> Either e a) -> (PQ.Oid, Int, Maybe ByteString) -> a+converter f (_, _, Just bs) = case f bs of+ Left err -> throw $ ConversionError (show err)+ Right val -> val+converter f (_, _, Nothing) = throw $ ConversionError "Excepted non-null"++-------------------------------------------------------------------------------+-- Byte Shuffling+-------------------------------------------------------------------------------++-- * Postgres varlena array has the following internal structure:+-- 7 * <vl_len_> - standard varlena header word+-- 8 * <ndim> - number of dimensions of the array+-- 9 * <dataoffset> - offset to stored data, or 0 if no nulls bitmap+-- 10 * <elemtype> - element type OID+-- 11 * <dimensions> - length of each array axis (C array of int)+-- 12 * <lower bnds> - lower boundary of each dimension (C array of int)+-- 13 * <null bitmap> - bitmap showing locations of nulls (OPTIONAL)+-- 14 * <actual data> - whatever is the stored data++-- Binary cursor overhead is 20 bytes.+calculateSize :: PQ.Oid -> Int -> Int+calculateSize ty len =+ case ty of+ PQ.Oid 16 -> 1 -- bool+ PQ.Oid 21 -> 2 -- int2+ PQ.Oid 23 -> 4 -- int4+ PQ.Oid 20 -> 8 -- int8+ PQ.Oid 700 -> 4 -- float4+ PQ.Oid 701 -> 8 -- float8+ PQ.Oid 25 -> -1 -- text+ PQ.Oid 2950 -> -1 -- uuid+ PQ.Oid 1007 -> (len - 20) `div` 8 -- int4[]+ PQ.Oid 1016 -> (len - 20) `div` 8 -- int8[]+ PQ.Oid 1021 -> (len - 20) `div` 8 -- float4[]+ _ -> error $ "Size not yet implemented" ++ (show ty)++-- 16 -> bool+-- 17 -> bytea+-- 18 -> char+-- 19 -> name+-- 20 -> int8+-- 21 -> int2+-- 23 -> int4+-- 24 -> regproc+-- 25 -> text+-- 26 -> oid+-- 27 -> tid+-- 28 -> xid+-- 29 -> cid+-- 700 -> float4+-- 701 -> float8+-- 702 -> abstime+-- 703 -> reltime+-- 704 -> tinterval+-- 705 -> unknown+-- 790 -> money+-- 1042 -> bpchar+-- 1043 -> varchar+-- 1082 -> date+-- 1083 -> time+-- 1114 -> timestamp+-- 1184 -> timestamptz+-- 1186 -> interval+-- 1266 -> timetz+-- 1560 -> bit+-- 1562 -> varbit+-- 1700 -> numeric+-- 2249 -> record+-- 2278 -> void+-- 2950 -> uuid++-------------------------------------------------------------------------------+-- Vector Extraction+-------------------------------------------------------------------------------++{-# INLINE extractor #-}+extractor :: ExFun -> ByteString -> Int -> IO (V.Vector a)+extractor f bs len = do+ vec <- VM.new len+ let (fptr, _, _) = toForeignPtr bs+ let (aptr, _) = VM.unsafeToForeignPtr0 vec++ rc <- withForeignPtr fptr $ \iptr ->+ withForeignPtr aptr $ \optr ->+ f iptr optr (fromIntegral len)++ -- XXX better error handling+ when (rc /= len) $ throwIO $ ConversionError "Extraction kernel malfunctioned."+ ovec <- V.unsafeFreeze vec+ -- CInt and Int look the same in memory, so coercion is safe+ return (unsafeCoerce ovec)++{-# NOINLINE float32vector #-}+float32vector :: ByteString -> Int -> IO (V.Vector Float)+float32vector = extractor extract_numeric_array++{-# NOINLINE int4vector #-}+int4vector :: ByteString -> Int -> IO (V.Vector Int32)+int4vector = extractor extract_int4_array++-------------------------------------------------------------------------------+-- Field Conversion+-------------------------------------------------------------------------------++-- | Result set field+field :: FromField a => RowParser a+field = fieldWith fromField++newtype RowParser a = RP { unRP :: ReaderT Row (State PQ.Column) a }+ deriving ( Functor, Applicative, Monad )++runRowParser :: RowParser a -> Row -> a+runRowParser parser rw = evalState (runReaderT (unRP parser) rw) 0++fieldWith :: FieldParser a -> RowParser a+fieldWith fieldP = RP $ do+ Row{..} <- ask+ column <- lift get+ lift (put (column + 1))+ let+ -- !ncols = PQ.nfields rowresult+ !result = rowresult+ !typeOid = unsafeDupablePerformIO $ PQ.ftype result column+ !len = unsafeDupablePerformIO $ PQ.getlength result row column+ !buffer = unsafeDupablePerformIO $ PQ.getvalue result row column+ return $ fieldP (typeOid, calculateSize typeOid len, buffer)++-------------------------------------------------------------------------------+-- Result Sets+-------------------------------------------------------------------------------++rowLoop :: (Ord n, Num n) => n -> n -> (n -> IO a) -> IO [a]+rowLoop lo hi m = loop hi []+ where+ loop !n !as+ | n < lo = return as+ | otherwise = do+ a <- m n+ loop (n-1) (a:as)++parseRows :: FromRow r => Query -> PQ.Result -> IO [r]+parseRows q result = do+ status <- liftIO $ PQ.resultStatus result+ case status of+ PQ.EmptyQuery ->+ liftIO $ throwIO $ QueryError "query: Empty query" q++ PQ.CommandOk -> do+ liftIO $ throwIO $ QueryError "query resulted in a command response" q++ PQ.TuplesOk -> do+ nrows <- liftIO $ PQ.ntuples result+ --ncols <- liftIO $ PQ.nfields result+ xs <- rowLoop 0 (nrows-1) $ \row -> do+ let rw = Row row result+ {-coltype <- liftIO $ PQ.ftype result c-}+ {-len <- liftIO $ PQ.getlength result row c-}+ {-size <- liftIO $ PQ.fsize result c-}+ {-buffer <- liftIO $ PQ.getvalue result row c-}+ return $ runRowParser fromRow rw+ return xs++ _ -> do+ err <- liftIO $ PQ.resultErrorMessage result+ liftIO $ throwIO $ QueryError (show err) q
+ src/Database/PostgreSQL/Stream/Parallel.hs view
@@ -0,0 +1,34 @@+module Database.PostgreSQL.Stream.Parallel (+ parallelStream,+) where++import Control.Monad.Trans+import Control.Concurrent+import Control.Concurrent.STM++import Data.Conduit+import Data.Conduit.TMChan+import Data.Conduit.TQueue+import Control.Monad.Trans.Resource (runResourceT, ResourceT)++import Database.PostgreSQL.Stream.Connection+import qualified Database.PostgreSQL.LibPQ as PQ++parallelStream ::+ PQ.Connection+ -> (PQ.Connection -> Source (ResourceT IO) a) -- Source+ -> Sink a (ResourceT IO) () -- Sink+ -> IO ()+parallelStream conn producer consumer = do+ chan <- atomically $ newTBMChan 32+ withPgConnection conn $ \c -> do++ tid <- forkIO . runResourceT+ $ producer c+ $$ sinkTBMChan chan True++ res <- runResourceT+ $ sourceTBMChan chan+ $$ consumer++ print res
+ src/Database/PostgreSQL/Stream/QueryBuilder.hs view
@@ -0,0 +1,187 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeSynonymInstances #-}++module Database.PostgreSQL.Stream.QueryBuilder (+ -- ** Quasiquoter+ sql,++ -- ** Query formatting+ fmtQuery,+ fmtSQL,++ -- ** Typeclasses+ ToSQL(..),+ ToField(..),+) where++import Database.PostgreSQL.Stream.Types++import Data.Int+import Data.Monoid+import Data.UUID as UUID+import Data.Text (Text)+import Data.Text.Encoding (encodeUtf8)++import Language.Haskell.TH+import Language.Haskell.TH.Quote++import Data.ByteString+import Data.ByteString.Search+import Data.ByteString.Lazy (toStrict)+import qualified Data.ByteString.Char8 as B8++-------------------------------------------------------------------------------+-- Arguments+-------------------------------------------------------------------------------++class ToField a where+ toField :: a -> Action++instance ToField Int where+ toField = Plain . B8.pack . show++instance ToField Int32 where+ toField = Plain . B8.pack . show++instance ToField Float where+ toField = Plain . B8.pack . show++instance ToField Double where+ toField = Plain . B8.pack . show++instance ToField ByteString where+ toField = Plain++instance ToField Integer where+ toField = Plain . B8.pack . show++instance ToField Char where+ toField = Plain . inQuotes . B8.pack . show++instance ToField String where+ toField = Plain . inQuotes . B8.pack++instance ToField Text where+ toField = Plain . inQuotes . encodeUtf8++-- SQL Identifier+instance ToField Identifier where+ toField = Plain . unIdentifier++-- SQL Expression+instance ToField SQL where+ toField = Plain . unSQL++-- Subquery (without substuttion, discards parameters)+instance ToField Query where+ toField (Query a) = Plain a++instance ToField UUID where+ toField = Plain . inQuotes . UUID.toASCIIBytes++instance ToField Null where+ toField _ = Plain "null"++instance (ToField a) => ToField (Only a) where+ toField (Only a) = toField a++instance (ToField a) => ToField (Maybe a) where+ toField Nothing = toField Null+ toField (Just a) = toField a++instance ToField Bool where+ toField True = Plain "true"+ toField False = Plain "false"++instance ToField Action where+ toField = id++inQuotes :: ByteString -> ByteString+inQuotes x = "\'" <> x <> "\'"++-------------------------------------------------------------------------------+-- ToSQL+-------------------------------------------------------------------------------++class ToSQL a where+ toSQL :: a -> (ByteString -> ByteString)++instance ToSQL () where+ toSQL _ = runFormatter []++instance (ToField a) => ToSQL (Only a) where+ toSQL (Only a) = runFormatter [toField a]++instance (ToField a) => ToSQL [a] where+ toSQL a = runFormatter (fmap toField a)++instance (ToField a) => ToSQL (Maybe a) where+ toSQL Nothing = runFormatter []+ toSQL (Just a) = runFormatter [toField a]++instance (ToField a, ToField b) => ToSQL (a,b) where+ toSQL (a,b) = runFormatter [toField a, toField b]++instance (ToField a, ToField b, ToField c) => ToSQL (a,b,c) where+ toSQL (a,b,c) = runFormatter [toField a, toField b, toField c]++instance (ToField a, ToField b, ToField c, ToField d) => ToSQL (a,b,c,d) where+ toSQL (a,b,c,d) = runFormatter [toField a, toField b, toField c, toField d]++instance (ToField a, ToField b, ToField c, ToField d, ToField e) => ToSQL (a,b,c,d,e) where+ toSQL (a,b,c,d,e) = runFormatter [toField a, toField b, toField c, toField d, toField e]++instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f) => ToSQL (a,b,c,d,e,f) where+ toSQL (a,b,c,d,e,f) = runFormatter [toField a, toField b, toField c, toField d, toField e, toField f]++instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f, ToField g) => ToSQL (a,b,c,d,e,f,g) where+ toSQL (a,b,c,d,e,f,g) = runFormatter [toField a, toField b, toField c, toField d, toField e, toField f, toField g]++instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f, ToField g, ToField h) => ToSQL (a,b,c,d,e,f,g,h) where+ toSQL (a,b,c,d,e,f,g,h) = runFormatter [toField a, toField b, toField c, toField d, toField e, toField f, toField g, toField h]++instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f, ToField g, ToField h, ToField i) => ToSQL (a,b,c,d,e,f,g,h,i) where+ toSQL (a,b,c,d,e,f,g,h,i) = runFormatter [toField a, toField b, toField c, toField d, toField e, toField f, toField g, toField h, toField i]++-------------------------------------------------------------------------------+-- Formatter+-------------------------------------------------------------------------------++render :: Action -> ByteString+render (Plain x) = x+render (Escape x) = error "Not implemented"+render (EscapeIdentifier x) = error "Not implemented"++sql :: QuasiQuoter+sql = QuasiQuoter+ { quotePat = error "Patterns are not supported"+ , quoteType = error "Types are not supported"+ , quoteExp = sqlExp+ , quoteDec = error "Declarations are not supported"+ }++sqlExp :: String -> Q Exp+sqlExp = stringE++-- Run the substitutions over a bytestring+runFormatter :: [Action] -> ByteString -> ByteString+runFormatter args input = loop args 1 input+ where+ loop (x:xs) i s = loop xs (i+1) $ toStrict (replace ("{" <> ix i <> "}") (render x) s)+ loop [] _ s = s++ ix :: Int -> ByteString+ ix = B8.pack . show++-------------------------------------------------------------------------------+-- query+-------------------------------------------------------------------------------++fmtQuery :: ToSQL a => Query -> a -> ByteString+fmtQuery q args = toSQL args (fromQuery q)++fmtSQL :: ToSQL a => Query -> a -> SQL+fmtSQL q args = SQLExpr $ toSQL args (fromQuery q)
+ src/Database/PostgreSQL/Stream/Types.hs view
@@ -0,0 +1,88 @@+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Database.PostgreSQL.Stream.Types (+ -- * Core types+ Identifier(..),+ SQL(..),+ Only(..),+ Null(..),+ Action(..),+ Query(..),++ -- * Exceptions+ QueryError(..),+ ConversionError(..),+) where++import Data.Monoid+import Data.String+import Data.Typeable+import Data.ByteString as B+import Data.ByteString.Char8 as B8++import Control.Exception++-------------------------------------------------------------------------------+-- Core Types+-------------------------------------------------------------------------------++-- | Literal SQL logic spliced in as a subexpression+newtype SQL = SQLExpr { unSQL :: ByteString }+ deriving (Eq, Ord, Show, IsString, Monoid)++-- | Newtype for a singular result set or argument value.+newtype Only a = Only { unOnly :: a }+ deriving (Eq, Ord, Read, Show, Typeable, Functor)++-- | Literal SQL identifier (i.e. table field names), spliced into the SQL query+-- unquoted.+newtype Identifier = Identifier { unIdentifier :: ByteString }+ deriving (Eq, Ord, Show, IsString, Monoid)++-- | SQL Null type+data Null = Null++-- | SQL Query subexpression+data Action+ = Plain ByteString+ | Escape ByteString+ | EscapeIdentifier ByteString+ deriving (Eq, Typeable, Show)++-- | SQL Query+newtype Query = Query { fromQuery :: ByteString }+ deriving (Eq, Ord, Typeable)++instance Show Query where+ show = show . fromQuery++instance IsString Query where+ fromString s = Query (B8.pack s)++instance Monoid Query where+ mempty = Query B.empty+ mappend (Query a) (Query b) = Query (B.append a b)+ {-# INLINE mappend #-}+ mconcat xs = Query (B.concat (fmap fromQuery xs))++-------------------------------------------------------------------------------+-- Failure+-------------------------------------------------------------------------------++data QueryError = QueryError+ { qeMessage :: String+ , qeQuery :: Query+ } deriving (Eq, Show, Typeable)++instance Exception QueryError++data ConversionError+ = ConversionError { ceMessage :: String } -- ^ Bytestring is malformed+ | UnexpectedNull { ceMessage :: String } -- ^ Unexpected null value+ | Incompatible { ceMessage :: String } -- ^ Type is incompatible with data+ deriving (Eq, Show, Typeable)++instance Exception ConversionError