diff --git a/Database/Cassandra/CQL.hs b/Database/Cassandra/CQL.hs --- a/Database/Cassandra/CQL.hs +++ b/Database/Cassandra/CQL.hs @@ -3,12 +3,9 @@ BangPatterns, OverlappingInstances, DataKinds, GADTs, KindSignatures #-} -- | Haskell client for Cassandra's CQL protocol -- --- This module isn't properly documented yet. For now, take a look at tests/example.hs. --- --- Credentials are not implemented yet. +-- For examples, take a look at the /tests/ directory in the source archive. -- --- Here's the correspondence between Haskell and CQL types. Not all Cassandra data --- types supported as yet: Haskell types listed below have been implemented. +-- Here's the correspondence between CQL and Haskell types: -- -- * ascii - 'ByteString' -- @@ -20,7 +17,7 @@ -- -- * counter - 'Counter' -- --- * decimal - 'Integer' +-- * decimal - 'Decimal' -- -- * double - 'Double' -- @@ -28,15 +25,13 @@ -- -- * int - 'Int' -- --- * text - 'Text' +-- * text / varchar - 'Text' -- -- * timestamp - 'UTCTime' -- -- * uuid - 'UUID' -- --- * varchar --- --- * varint +-- * varint - 'Integer' -- -- * timeuuid - 'TimeUUID' -- @@ -48,44 +43,92 @@ -- -- * set\ - 'Set' b -- +-- The recommended way to use it is to specify your queries as global constants +-- in this way: +-- +-- > createSongs :: Query Schema () () +-- > createSongs = "create table songs (id uuid PRIMARY KEY, title text, artist text, comment text)" +-- > +-- > insertSong :: Query Write (UUID, Text, Text, Maybe Text) () +-- > insertSong = "insert into songs (id, title, artist, comment) values (?, ?, ?)" +-- > +-- > getOneSong :: Query Rows UUID (Text, Text, Maybe Text) +-- > getOneSong = "select title, artist, comment from songs where id=?" +-- +-- The three type parameters are the query type ('Schema', 'Write' or 'Rows') followed by the +-- input and output types, which are given as tuples whose constituent types must match +-- the ones in the query CQL. If you do not match them correctly, you'll get a runtime +-- error when you execute the query. If you do, then the query becomes completely type +-- safe. +-- +-- Types can be 'Maybe' types, in which case you can read and write a Cassandra \'null\' +-- in the table. Cassandra allows any column to be null, but you can lock this out by +-- specifying non-Maybe types. +-- +-- The reason why it is desirable for 'Query' values to be global constants (i.e. defined +-- at top-level of your Haskell source) is because it's the fastest at runtime: +-- Queries contain a hash value that is calculated lazily from the query text, which is +-- used to locally cache prepared queries. If the constant is global, then Haskell caches +-- the hash value so it gets calculated once only per query per program execution. +-- +-- The query types are: +-- +-- * 'Schema' for modifications to the schema. The output tuple type must be (). +-- +-- * 'Write' for row inserts and updates, and such. The output tuple type must be (). +-- +-- * 'Rows' for selects that give a list of rows in response. +-- +-- The functions to use for these query types are 'executeSchema', 'executeWrite' and +-- 'executeRows' or 'executeRow' respectively. +-- +-- /To do/ +-- +-- * Add credentials. +-- +-- * Improve connection pooling. +-- +-- * Add the ability to easily run queries in parallel. + module Database.Cassandra.CQL ( -- * Initialization Server, - createCassandraPool, - CPool, - -- * Monads + Keyspace(..), + Pool, + newPool, + -- * Cassandra monad MonadCassandra, Cas, runCas, CassandraException(..), + CassandraCommsError(..), TransportDirection(..), - -- * Type used in operations - Keyspace(..), - Result(..), - TableSpec(..), - ColumnSpec(..), - Metadata(..), - CType(..), - Change(..), - Table(..), - Consistency(..), -- * Queries Query, - query, Style(..), - -- * Operations + query, + -- * Executing queries + Consistency(..), + Change(..), + executeSchema, + executeWrite, executeRows, executeRow, - executeWrite, - executeSchema, - executeRaw, -- * Value types CasType(..), CasValues(..), Blob(..), Counter(..), TimeUUID(..), - metadataTypes + metadataTypes, + -- * Lower-level interfaces + executeRaw, + Result(..), + TableSpec(..), + ColumnSpec(..), + Metadata(..), + CType(..), + Table(..) ) where import Control.Applicative @@ -103,6 +146,7 @@ import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L import Data.Data +import Data.Decimal import Data.Either (lefts) import Data.Int import Data.List @@ -145,24 +189,26 @@ sesActive :: Maybe ActiveSession } -data CPool = CPool { +-- | A handle for the state of the connection pool. +data Pool = Pool { piKeyspace :: Keyspace, piSessions :: TVar (Seq Session) } class MonadCatchIO m => MonadCassandra m where - getEltsandraPool :: m CPool + getEltsandraPool :: m Pool -createCassandraPool :: [Server] -> Keyspace -> IO CPool -createCassandraPool svrs ks = do +-- | Construct a pool of Cassandra connections. +newPool :: [Server] -> Keyspace -> IO Pool +newPool svrs ks = do let sessions = map (\svr -> Session svr Nothing) svrs sess <- atomically $ newTVar (Seq.fromList sessions) - return $ CPool { + return $ Pool { piKeyspace = ks, piSessions = sess } -takeSession :: CPool -> IO Session +takeSession :: Pool -> IO Session takeSession pool = atomically $ do sess <- readTVar (piSessions pool) if Seq.null sess @@ -172,10 +218,10 @@ writeTVar (piSessions pool) (Seq.drop 1 sess) return ses -putSession :: CPool -> Session -> IO () +putSession :: Pool -> Session -> IO () putSession pool ses = atomically $ modifyTVar (piSessions pool) (|> ses) -connectIfNeeded :: CPool -> Session -> IO Session +connectIfNeeded :: Pool -> Session -> IO Session connectIfNeeded pool session = if isJust (sesActive session) then return session @@ -272,7 +318,7 @@ recvAll :: Socket -> Int -> IO ByteString recvAll s n = do bs <- recv s n - when (B.null bs) $ throw $ LocalProtocolError $ "short read" + when (B.null bs) $ throw ShortRead let left = n - B.length bs if left == 0 then return bs @@ -298,6 +344,7 @@ else liftIO $ recvAll s (fromIntegral length) --liftIO $ putStrLn $ hexdump 0 (C.unpack $ hdrBs `B.append` body) return $ Frame flags stream opcode body + `catch` \exc -> throw $ CassandraIOException exc where parseHeader = do ver <- getWord8 @@ -319,16 +366,12 @@ --liftIO $ putStrLn $ hexdump 0 (C.unpack bs) s <- gets actSocket liftIO $ sendAll s bs + `catch` \exc -> throw $ CassandraIOException exc class ProtoElt a where getElt :: Get a putElt :: a -> Put -class CasType a where - getCas :: Get a - putCas :: a -> Put - casType :: a -> CType - encodeElt :: ProtoElt a => a -> ByteString encodeElt = runPut . putElt @@ -400,11 +443,10 @@ Long <$> getByteString (fromIntegral len) data TransportDirection = TransportSending | TransportReceiving - deriving Show + deriving (Eq, Show) -data CassandraException = LocalProtocolError Text - | AuthenticationException Text - | ValueMarshallingException TransportDirection Text +-- | An exception that indicates an error originating in the Cassandra server. +data CassandraException = AuthenticationException Text | ServerError Text | ProtocolError Text | BadCredentials Text @@ -424,6 +466,22 @@ instance Exception CassandraException where +-- | All errors at the communications level are reported with this exception +-- ('IOException's from socket I/O are always wrapped), and this exception +-- typically would mean that a retry is warranted. +-- +-- Note that this exception isn't guaranteed to be a transient one, so a limit +-- on the number of retries is likely to be a good idea. +-- 'LocalProtocolError' probably indicates a corrupted database or driver +-- bug. +data CassandraCommsError = LocalProtocolError Text + | ValueMarshallingException TransportDirection Text + | CassandraIOException IOException + | ShortRead + deriving (Show, Typeable) + +instance Exception CassandraCommsError + throwError :: MonadCatchIO m => ByteString -> m a throwError bs = do case runGet parseError bs of @@ -459,7 +517,7 @@ 0x2500 -> Unprepared <$> getElt <*> getElt _ -> fail $ "unknown error code 0x"++showHex code "" -introduce :: CPool -> StateT ActiveSession IO () +introduce :: Pool -> StateT ActiveSession IO () introduce pool = do sendFrame $ Frame [] 0 STARTUP $ encodeElt $ ([("CQL_VERSION", "3.0.0")] :: [(Text, Text)]) fr <- recvFrame @@ -476,7 +534,7 @@ _ -> throw $ ProtocolError $ "expected SetKeyspace, but got " `T.append` T.pack (show res) `T.append` " for query: " `T.append` T.pack (show q) -withSession :: MonadCassandra m => (CPool -> StateT ActiveSession IO a) -> m a +withSession :: MonadCassandra m => (Pool -> StateT ActiveSession IO a) -> m a withSession code = do pool <- getEltsandraPool mA <- liftIO $ do @@ -503,25 +561,32 @@ Just a -> return a Nothing -> withSession code -- Try again until we succeed +-- | The name of a Cassandra keyspace. See the Cassandra documentation for more +-- information. newtype Keyspace = Keyspace Text deriving (Eq, Ord, Show, IsString, ProtoElt) +-- | The name of a Cassandra table (a.k.a. column family). newtype Table = Table Text deriving (Eq, Ord, Show, IsString, ProtoElt) +-- | A fully qualified identification of a table that includes the 'Keyspace'. data TableSpec = TableSpec Keyspace Table - deriving Show + deriving (Eq, Ord, Show) instance ProtoElt TableSpec where putElt _ = error "formatting TableSpec is not implemented" getElt = TableSpec <$> getElt <*> getElt +-- | Information about a table column. data ColumnSpec = ColumnSpec TableSpec Text CType deriving Show +-- | The specification of a list of result set columns. data Metadata = Metadata [ColumnSpec] deriving Show +-- | Cassandra data types as used in metadata. data CType = CCustom Text | CAscii | CBigint @@ -535,15 +600,64 @@ | CText | CTimestamp | CUuid - | CVarchar | CVarint | CTimeuuid | CInet | CList CType | CMap CType CType | CSet CType - deriving (Eq, Ord, Show) + | CMaybe CType + deriving (Eq) +instance Show CType where + show ct = case ct of + CCustom name -> T.unpack name + CAscii -> "ascii" + CBigint -> "bigint" + CBlob -> "blob" + CBoolean -> "boolean" + CCounter -> "counter" + CDecimal -> "decimal" + CDouble -> "double" + CFloat -> "float" + CInt -> "int" + CText -> "text" + CTimestamp -> "timestamp" + CUuid -> "uuid" + CVarint -> "varint" + CTimeuuid -> "timeuuid" + CInet -> "inet" + CList t -> "list<"++show t++">" + CMap t1 t2 -> "map<"++show t1++","++show t2++">" + CSet t -> "set<"++show t++">" + CMaybe t -> "nullable "++show t + +equivalent :: CType -> CType -> Bool +equivalent (CMaybe a) (CMaybe b) = a == b +equivalent (CMaybe a) b = a == b +equivalent a (CMaybe b) = a == b +equivalent a b = a == b + +-- | A type class for types that can be used in query arguments or column values in +-- returned results. +class CasType a where + getCas :: Get a + putCas :: a -> Put + casType :: a -> CType + casNothing :: a + casNothing = error "casNothing impossible" + casObliterate :: a -> ByteString -> Maybe ByteString + casObliterate _ bs = Just bs + +instance CasType a => CasType (Maybe a) where + getCas = Just <$> getCas + putCas Nothing = return () + putCas (Just a) = putCas a + casType _ = CMaybe (casType (undefined :: a)) + casNothing = Nothing + casObliterate (Just a) bs = Just bs + casObliterate Nothing _ = Nothing + instance CasType ByteString where getCas = getByteString =<< remaining putCas = putByteString @@ -554,6 +668,8 @@ putCas = putWord64be . fromIntegral casType _ = CBigint +-- | If you wrap this round a 'ByteString', it will be treated as a /blob/ type +-- instead of /ascii/ (if it was a plain 'ByteString' type). newtype Blob = Blob ByteString deriving (Eq, Ord, Show) @@ -568,6 +684,7 @@ putCas False = putWord8 0 casType _ = CBoolean +-- | A Cassandra distributed counter value. newtype Counter = Counter Int64 deriving (Eq, Ord, Show, Read) @@ -589,20 +706,32 @@ else i putCas i = putByteString . B.pack $ if i < 0 - then encode 0x100 $ positivize 0x80 i - else encode 0x80 i + then encodeNeg $ positivize 0x80 i + else encodePos i where - encode :: Word8 -> Integer -> [Word8] - encode limit i = reverse $ enc i + encodePos :: Integer -> [Word8] + encodePos i = reverse $ enc i where - limit' = fromIntegral limit - enc i | i == 0 = [] - enc i | i < limit' = [fromIntegral i] - enc i = fromIntegral i : enc (i `shiftR` 8) + enc i | i == 0 = [0] + enc i | i < 0x80 = [fromIntegral i] + enc i = fromIntegral i : enc (i `shiftR` 8) + encodeNeg :: Integer -> [Word8] + encodeNeg i = reverse $ enc i + where + enc i | i == 0 = [] + enc i | i < 0x100 = [fromIntegral i] + enc i = fromIntegral i : enc (i `shiftR` 8) positivize :: Integer -> Integer -> Integer positivize bits i = case bits + i of i' | i' >= 0 -> i' + bits _ -> positivize (bits `shiftL` 8) i + casType _ = CVarint + +instance CasType Decimal where + getCas = Decimal <$> (fromIntegral . min 0xff <$> getWord32be) <*> getCas + putCas (Decimal places mantissa) = do + putWord32be (fromIntegral places) + putCas mantissa casType _ = CDecimal instance CasType Double where @@ -648,7 +777,9 @@ putCas = putByteString . L.toStrict . UUID.toByteString casType _ = CUuid -data TimeUUID = TimeUUID UUID deriving (Eq, Data, Ord, Read, Show, Typeable) +-- | If you wrap this round a 'UUID' then it is treated as a /timeuuid/ type instead of +-- /uuid/ (if it was a plain 'UUID' type). +newtype TimeUUID = TimeUUID UUID deriving (Eq, Data, Ord, Read, Show, Typeable) instance CasType TimeUUID where getCas = TimeUUID <$> getCas @@ -740,7 +871,9 @@ 0x0007 -> pure CDouble 0x0008 -> pure CFloat 0x0009 -> pure CInt - 0x000a -> pure CVarchar + --0x000a -> pure CVarchar -- Server seems to use CText even when 'varchar' is specified + -- i.e. they're interchangeable in the CQL and always + -- 'text' in the protocol. 0x000b -> pure CTimestamp 0x000c -> pure CUuid 0x000d -> pure CText @@ -772,8 +905,16 @@ newtype QueryID = QueryID (Digest SHA1) deriving (Eq, Ord, Show) -data Style = Rows | Write | Schema +-- | The first type argument for Query. Tells us what kind of query it is. +data Style = Schema -- ^ A query that modifies the schema, such as DROP TABLE or CREATE TABLE + | Write -- ^ A query that writes data, such as an INSERT or UPDATE + | Rows -- ^ A query that returns a list of rows, such as SELECT +-- | The text of a CQL query, along with type parameters to make the query type safe. +-- The type arguments are 'Style', followed by input and output column types for the +-- query each represented as a tuple. +-- +-- The /DataKinds/ language extension is required for 'Style'. data Query :: Style -> * -> * -> * where Query :: QueryID -> Text -> Query style i o deriving Show @@ -781,6 +922,14 @@ instance IsString (Query style i o) where fromString = query . T.pack +-- | Construct a query. Another way to construct one is as an overloaded string through +-- the 'IsString' instance if you turn on the /OverloadedStrings/ language extension, e.g. +-- +-- > {-# LANGUAGE OverloadedStrings #-} +-- > ... +-- > +-- > getOneSong :: Query Rows UUID (Text, Text, Maybe Text) +-- > getOneSong = "select title, artist, comment from songs where id=?" query :: Text -> Query style i o query cql = Query (QueryID . hash . T.encodeUtf8 $ cql) cql @@ -800,6 +949,7 @@ "DROPPED" -> pure DROPPED _ -> fail $ "unexpected change string: "++show str +-- | A low-level query result used with 'executeRaw'. data Result vs = Void | RowsResult Metadata [vs] | SetKeyspace Text @@ -814,7 +964,7 @@ f `fmap` Prepared pqid meta = Prepared pqid meta f `fmap` SchemaChange ch ks t = SchemaChange ch ks t -instance ProtoElt (Result [ByteString]) where +instance ProtoElt (Result [Maybe ByteString]) where putElt _ = error "formatting RESULT is not implemented" getElt = do kind <- getWord32be @@ -824,7 +974,11 @@ meta@(Metadata colSpecs) <- getElt let colCount = length colSpecs rowCount <- fromIntegral <$> getWord32be - rows <- replicateM rowCount (replicateM colCount (unLong <$> getElt)) + rows <- replicateM rowCount $ replicateM colCount $ do + len <- getWord32be + if len == 0xffffffff + then return Nothing + else Just <$> getByteString (fromIntegral len) return $ RowsResult meta rows 0x0003 -> SetKeyspace <$> getElt 0x0004 -> Prepared <$> getElt <*> getElt @@ -842,7 +996,7 @@ case frOpcode fr of RESULT -> do res <- decodeEltM "RESULT" (frBody fr) - case (res :: Result [ByteString]) of + case (res :: Result [Maybe ByteString]) of Prepared pqid meta -> do let pq = PreparedQuery pqid meta modify $ \act -> act { actQueryCache = M.insert qid pq (actQueryCache act) } @@ -854,15 +1008,17 @@ data CodingFailure = Mismatch Int CType CType | WrongNumber Int Int | DecodeFailure Int String + | NullValue Int CType instance Show CodingFailure where show (Mismatch i t1 t2) = "at value index "++show (i+1)++", Haskell type specifies "++show t1++", but database metadata says "++show t2 show (WrongNumber i1 i2) = "wrong number of values: Haskell type specifies "++show i1++" but database metadata says "++show i2 show (DecodeFailure i why) = "failed to decode value index "++show (i+1)++": "++why + show (NullValue i t) = "at value index "++show (i+1)++" received a null "++show t++" value but Haskell type is not a Maybe" class CasNested v where - encodeNested :: Int -> v -> [CType] -> Either CodingFailure [ByteString] - decodeNested :: Int -> [(CType, ByteString)] -> Either CodingFailure v + encodeNested :: Int -> v -> [CType] -> Either CodingFailure [Maybe ByteString] + decodeNested :: Int -> [(CType, Maybe ByteString)] -> Either CodingFailure v countNested :: v -> Int instance CasNested () where @@ -873,27 +1029,31 @@ countNested _ = 0 instance (CasType a, CasNested rem) => CasNested (a, rem) where - encodeNested !i (a, rem) (ta:trem) | ta == casType a = + encodeNested !i (a, rem) (ta:trem) | ta `equivalent` casType a = case encodeNested (i+1) rem trem of Left err -> Left err - Right brem -> Right $ encodeCas a : brem + Right brem -> Right $ ba : brem + where + ba = casObliterate a . encodeCas $ a encodeNested !i (a, _) (ta:_) = Left $ Mismatch i (casType a) ta encodeNested !i vs [] = Left $ WrongNumber (i + countNested vs) i - decodeNested !i ((ta, ba):rem) | ta == casType (undefined :: a) = - case decodeCas ba of - Left err -> Left $ DecodeFailure i err - Right a -> - case decodeNested (i+1) rem of - Left err -> Left err - Right arem -> Right (a, arem) + decodeNested !i ((ta, mba):rem) | ta `equivalent` casType (undefined :: a) = + case (decodeCas <$> mba, casType (undefined :: a), decodeNested (i+1) rem) of + (Nothing, CMaybe _, Right arem) -> Right (casNothing, arem) + (Nothing, _, _) -> Left $ NullValue i ta + (Just (Left err), _, _) -> Left $ DecodeFailure i err + (_, _, Left err) -> Left err + (Just (Right a), _, Right arem) -> Right (a, arem) decodeNested !i ((ta, _):rem) = Left $ Mismatch i (casType (undefined :: a)) ta decodeNested !i [] = Left $ WrongNumber (i + 1 + countNested (undefined :: rem)) i countNested _ = let n = 1 + countNested (undefined :: rem) in seq n n +-- | A type class for a tuple of 'CasType' instances, representing either a list of +-- arguments for a query, or the values in a row of returned query results. class CasValues v where - encodeValues :: v -> [CType] -> Either CodingFailure [ByteString] - decodeValues :: [(CType, ByteString)] -> Either CodingFailure v + encodeValues :: v -> [CType] -> Either CodingFailure [Maybe ByteString] + decodeValues :: [(CType, Maybe ByteString)] -> Either CodingFailure v instance CasValues () where encodeValues () types = encodeNested 0 () types @@ -1050,6 +1210,7 @@ decodeValues vs = (\(a, (b, (c, (d, (e, (f, (g, (h, (i, (j, (k, (l, (m, (n, (o, (p, (q, (r, (s, (t, ())))))))))))))))))))) -> (a, b, c, d, e, f, g, h, i, j, k, l, m, n, o, p, q, r, s, t)) <$> decodeNested 0 vs +-- | Cassandra consistency level. See the Cassandra documentation for an explanation. data Consistency = ANY | ONE | TWO | THREE | QUORUM | ALL | LOCAL_QUORUM | EACH_QUORUM deriving (Eq, Ord, Show, Bounded, Enum) @@ -1076,12 +1237,13 @@ 0x0007 -> pure EACH_QUORUM _ -> fail $ "unknown consistency value 0x"++showHex w "" +-- | A low-level function in case you need some rarely-used capabilities. executeRaw :: (MonadCassandra m, CasValues i) => - Query style any_i any_o -> i -> Consistency -> m (Result [ByteString]) + Query style any_i any_o -> i -> Consistency -> m (Result [Maybe ByteString]) executeRaw query i cons = withSession $ \_ -> executeInternal query i cons executeInternal :: CasValues values => - Query style any_i any_o -> values -> Consistency -> StateT ActiveSession IO (Result [ByteString]) + Query style any_i any_o -> values -> Consistency -> StateT ActiveSession IO (Result [Maybe ByteString]) executeInternal query i cons = do pq@(PreparedQuery pqid queryMeta) <- prepare query values <- case encodeValues i (metadataTypes queryMeta) of @@ -1090,10 +1252,13 @@ sendFrame $ Frame [] 0 EXECUTE $ runPut $ do putElt pqid putWord16be (fromIntegral $ length values) - forM_ values $ \value -> do - let enc = encodeCas value - putWord32be (fromIntegral $ B.length enc) - putByteString enc + forM_ values $ \mValue -> + case mValue of + Nothing -> putWord32be 0xffffffff + Just value -> do + let enc = encodeCas value + putWord32be (fromIntegral $ B.length enc) + putByteString enc putElt cons fr <- recvFrame case frOpcode fr of @@ -1101,9 +1266,12 @@ ERROR -> throwError (frBody fr) _ -> throw $ LocalProtocolError $ "execute: unexpected opcode " `T.append` T.pack (show (frOpcode fr)) --- | Execute a query that returns rows +-- | Execute a query that returns rows. executeRows :: (MonadCassandra m, CasValues i, CasValues o) => - Consistency -> Query Rows i o -> i -> m [o] + Consistency + -> Query Rows i o -- ^ CQL query to execute + -> i -- ^ Input values substituted in the query + -> m [o] executeRows cons q i = do res <- executeRaw q i cons case res of @@ -1114,12 +1282,15 @@ -- | Helper for 'executeRows' useful in situations where you are only expecting one row -- to be returned. executeRow :: (MonadCassandra m, CasValues i, CasValues o) => - Consistency -> Query Rows i o -> i -> m (Maybe o) + Consistency + -> Query Rows i o -- ^ CQL query to execute + -> i -- ^ Input values substituted in the query + -> m (Maybe o) executeRow cons q i = do rows <- executeRows cons q i return $ listToMaybe rows -decodeRows :: (MonadCatchIO m, CasValues values) => Metadata -> [[ByteString]] -> m [values] +decodeRows :: (MonadCatchIO m, CasValues values) => Metadata -> [[Maybe ByteString]] -> m [values] decodeRows meta rows0 = do let rows1 = flip map rows0 $ \cols -> decodeValues (zip (metadataTypes meta) cols) case lefts rows1 of @@ -1128,9 +1299,12 @@ let rows2 = flip map rows1 $ \(Right v) -> v return $ rows2 --- | Execute a write operation that returns void +-- | Execute a write operation that returns void. executeWrite :: (MonadCassandra m, CasValues i) => - Consistency -> Query Write i () -> i -> m () + Consistency + -> Query Write i () -- ^ CQL query to execute + -> i -- ^ Input values substituted in the query + -> m () executeWrite cons q i = do res <- executeRaw q i cons case res of @@ -1140,7 +1314,10 @@ -- | Execute a schema change, such as creating or dropping a table. executeSchema :: (MonadCassandra m, CasValues i) => - Consistency -> Query Schema i () -> i -> m (Change, Keyspace, Table) + Consistency + -> Query Schema i () -- ^ CQL query to execute + -> i -- ^ Input values substituted in the query + -> m (Change, Keyspace, Table) executeSchema cons q i = do res <- executeRaw q i cons case res of @@ -1148,15 +1325,18 @@ _ -> throw $ ProtocolError $ "expected SchemaChange, but got " `T.append` T.pack (show res) `T.append` " for query: " `T.append` T.pack (show q) +-- | A helper for extracting the types from a metadata definition. metadataTypes :: Metadata -> [CType] metadataTypes (Metadata colspecs) = map (\(ColumnSpec _ _ typ) -> typ) colspecs -newtype Cas a = Cas (ReaderT CPool IO a) +-- | The monad used to run Cassandra queries in. +newtype Cas a = Cas (ReaderT Pool IO a) deriving (Functor, Applicative, Monad, MonadIO, MonadCatchIO) instance MonadCassandra Cas where getEltsandraPool = Cas ask -runCas :: CPool -> Cas a -> IO a +-- | Execute Cassandra queries. +runCas :: Pool -> Cas a -> IO a runCas pool (Cas code) = runReaderT code pool diff --git a/cassandra-cql.cabal b/cassandra-cql.cabal --- a/cassandra-cql.cabal +++ b/cassandra-cql.cabal @@ -1,5 +1,5 @@ name: cassandra-cql -version: 0.1.0.0 +version: 0.2.0.0 synopsis: Haskell client for Cassandra's CQL protocol description: Haskell client for Cassandra's CQL protocol @@ -22,6 +22,7 @@ tests/test-decimal.hs tests/test-timeuuid.hs tests/test-inet.hs + tests/test-varint.hs source-repository head type: git @@ -40,7 +41,8 @@ cryptohash >= 0.9.0, stm >= 2.4.0, uuid >= 1.2.0, - time >= 1.4.0.0 + time >= 1.4.0.0, + Decimal >= 0.3.0 ghc-options: -Wall -fno-warn-name-shadowing -fno-warn-unused-matches -fno-warn-missing-signatures -fno-warn-orphans -fno-warn-unused-imports -fno-warn-unused-binds diff --git a/tests/example.hs b/tests/example.hs --- a/tests/example.hs +++ b/tests/example.hs @@ -4,7 +4,6 @@ import Control.Monad import Control.Monad.CatchIO import Control.Monad.Trans (liftIO) -import Data.Int import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as C import Data.Text (Text) @@ -16,13 +15,13 @@ dropSongs = "drop table songs" createSongs :: Query Schema () () -createSongs = "create table songs (id uuid PRIMARY KEY, title ascii, artist text, femaleSinger boolean, timesPlayed int)" +createSongs = "create table songs (id uuid PRIMARY KEY, title ascii, artist varchar, femaleSinger boolean, timesPlayed int, comment text)" -insertSong :: Query Write (UUID, ByteString, Text, Bool, Int) () -insertSong = "insert into songs (id, title, artist, femaleSinger, timesPlayed) values (?, ?, ?, ?, ?)" +insertSong :: Query Write (UUID, ByteString, Text, Bool, Int, Maybe Text) () +insertSong = "insert into songs (id, title, artist, femaleSinger, timesPlayed, comment) values (?, ?, ?, ?, ?, ?)" -getSongs :: Query Rows () (UUID, ByteString, Text, Bool, Int) -getSongs = "select id, title, artist, femaleSinger, timesPlayed from songs" +getSongs :: Query Rows () (UUID, ByteString, Text, Bool, Int, Maybe Text) +getSongs = "select id, title, artist, femaleSinger, timesPlayed, comment from songs" getOneSong :: Query Rows UUID (Text, Int) getOneSong = "select artist, timesPlayed from songs where id=?" @@ -37,7 +36,7 @@ Assuming a 'test' keyspace already exists. Here's some CQL to create it: CREATE KEYSPACE test WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : '1' }; -} - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropSongs () liftIO . print =<< executeSchema QUORUM createSongs () @@ -45,18 +44,19 @@ u1 <- liftIO randomIO u2 <- liftIO randomIO u3 <- liftIO randomIO - executeWrite QUORUM insertSong (u1, "La Grange", "ZZ Top", False, 2) - executeWrite QUORUM insertSong (u2, "Your Star", "Evanescence", True, 799) - executeWrite QUORUM insertSong (u3, "Angel of Death", "Slayer", False, 50) + executeWrite QUORUM insertSong (u1, "La Grange", "ZZ Top", False, 2, Nothing) + executeWrite QUORUM insertSong (u2, "Your Star", "Evanescence", True, 799, Nothing) + executeWrite QUORUM insertSong (u3, "Angel of Death", "Slayer", False, 50, Just "Singer Tom Araya") songs <- executeRows QUORUM getSongs () - liftIO $ forM_ songs $ \(uuid, title, artist, female, played) -> do + liftIO $ forM_ songs $ \(uuid, title, artist, female, played, mComment) -> do putStrLn "" putStrLn $ "id : "++show uuid putStrLn $ "title : "++C.unpack title putStrLn $ "artist : "++T.unpack artist putStrLn $ "female singer : "++show female putStrLn $ "times played : "++show played + putStrLn $ "comment : "++show mComment liftIO $ putStrLn "" liftIO . print =<< executeRow QUORUM getOneSong u2 diff --git a/tests/test-decimal.hs b/tests/test-decimal.hs --- a/tests/test-decimal.hs +++ b/tests/test-decimal.hs @@ -7,7 +7,7 @@ import Data.Int import Data.Text (Text) import qualified Data.Text as T -import Data.Time.Clock +import Data.Decimal import Data.UUID import System.Random @@ -17,10 +17,10 @@ createLists :: Query Schema () () createLists = "create table decimals (id uuid PRIMARY KEY, item decimal)" -insert :: Query Write (UUID, Integer) () +insert :: Query Write (UUID, Decimal) () insert = "insert into decimals (id, item) values (?, ?)" -select :: Query Rows UUID Integer +select :: Query Rows UUID Decimal select = "select item from decimals where id=?" ignoreDropFailure :: Cas () -> Cas () @@ -29,7 +29,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -42,20 +42,16 @@ u6 <- liftIO randomIO u7 <- liftIO randomIO u8 <- liftIO randomIO - u9 <- liftIO randomIO - u10 <- liftIO randomIO - executeWrite QUORUM insert (u1, 0) - executeWrite QUORUM insert (u2, -1) - executeWrite QUORUM insert (u3, 12345678901234567890123456789) - executeWrite QUORUM insert (u4, -12345678901234567890123456789) - executeWrite QUORUM insert (u5, -65537) - executeWrite QUORUM insert (u6, -65536) - executeWrite QUORUM insert (u7, -65535) - executeWrite QUORUM insert (u8, -32769) - executeWrite QUORUM insert (u9, -32768) - executeWrite QUORUM insert (u10, -32767) + executeWrite QUORUM insert (u1, read "0") + executeWrite QUORUM insert (u2, read "1.02") + executeWrite QUORUM insert (u3, read "12345678901234567890.123456789") + executeWrite QUORUM insert (u4, read "-12345678901234567890.123456789") + executeWrite QUORUM insert (u5, read "3.141592654") + executeWrite QUORUM insert (u6, read "-3.141592654") + executeWrite QUORUM insert (u7, read "-0.000000001") + executeWrite QUORUM insert (u8, read "118000") - let us = [u1,u2,u3,u4,u5,u6,u7,u8,u9,u10] + let us = [u1,u2,u3,u4,u5,u6,u7, u8] forM_ us $ \u -> liftIO . print =<< executeRow QUORUM select u diff --git a/tests/test-double.hs b/tests/test-double.hs --- a/tests/test-double.hs +++ b/tests/test-double.hs @@ -32,7 +32,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-float.hs b/tests/test-float.hs --- a/tests/test-float.hs +++ b/tests/test-float.hs @@ -32,7 +32,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-inet.hs b/tests/test-inet.hs --- a/tests/test-inet.hs +++ b/tests/test-inet.hs @@ -31,7 +31,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-list.hs b/tests/test-list.hs --- a/tests/test-list.hs +++ b/tests/test-list.hs @@ -42,7 +42,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropListsI () diff --git a/tests/test-set.hs b/tests/test-set.hs --- a/tests/test-set.hs +++ b/tests/test-set.hs @@ -32,7 +32,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-timestamp.hs b/tests/test-timestamp.hs --- a/tests/test-timestamp.hs +++ b/tests/test-timestamp.hs @@ -29,7 +29,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-timeuuid.hs b/tests/test-timeuuid.hs --- a/tests/test-timeuuid.hs +++ b/tests/test-timeuuid.hs @@ -35,7 +35,7 @@ _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () diff --git a/tests/test-varint.hs b/tests/test-varint.hs new file mode 100644 --- /dev/null +++ b/tests/test-varint.hs @@ -0,0 +1,61 @@ +{-# LANGUAGE OverloadedStrings, DataKinds #-} + +import Database.Cassandra.CQL +import Control.Monad +import Control.Monad.CatchIO +import Control.Monad.Trans (liftIO) +import Data.Int +import Data.Text (Text) +import qualified Data.Text as T +import Data.Time.Clock +import Data.UUID +import System.Random + +dropLists :: Query Schema () () +dropLists = "drop table varints" + +createLists :: Query Schema () () +createLists = "create table varints (id uuid PRIMARY KEY, item varint)" + +insert :: Query Write (UUID, Integer) () +insert = "insert into varints (id, item) values (?, ?)" + +select :: Query Rows UUID Integer +select = "select item from varints where id=?" + +ignoreDropFailure :: Cas () -> Cas () +ignoreDropFailure code = code `catch` \exc -> case exc of + ConfigError _ -> return () -- Ignore the error if the table doesn't exist + _ -> throw exc + +main = do + pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + runCas pool $ do + ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () + liftIO . print =<< executeSchema QUORUM createLists () + + u1 <- liftIO randomIO + u2 <- liftIO randomIO + u3 <- liftIO randomIO + u4 <- liftIO randomIO + u5 <- liftIO randomIO + u6 <- liftIO randomIO + u7 <- liftIO randomIO + u8 <- liftIO randomIO + u9 <- liftIO randomIO + u10 <- liftIO randomIO + executeWrite QUORUM insert (u1, 0) + executeWrite QUORUM insert (u2, -1) + executeWrite QUORUM insert (u3, 12345678901234567890123456789) + executeWrite QUORUM insert (u4, -12345678901234567890123456789) + executeWrite QUORUM insert (u5, -65537) + executeWrite QUORUM insert (u6, -65536) + executeWrite QUORUM insert (u7, -65535) + executeWrite QUORUM insert (u8, -32769) + executeWrite QUORUM insert (u9, -32768) + executeWrite QUORUM insert (u10, -32767) + + let us = [u1,u2,u3,u4,u5,u6,u7,u8,u9,u10] + forM_ us $ \u -> + liftIO . print =<< executeRow QUORUM select u +