diff --git a/cassy.cabal b/cassy.cabal --- a/cassy.cabal +++ b/cassy.cabal @@ -1,16 +1,19 @@ Name: cassy -Version: 0.4.0.1 +Version: 0.5.1.0 Synopsis: A high level driver for the Cassandra datastore License: BSD3 License-file: LICENSE Author: Ozgun Ataman Maintainer: ozataman@gmail.com Homepage: http://github.com/ozataman/cassy -Category: Data +Category: Database Build-type: Simple description: The objective is to completely isolate away the thrift layer, providing - a more idiomatic Haskell experience working with Cassandra. + a more idiomatic Haskell experience working with Cassandra. Be sure + to check out the README on Github for some more explanation and + Release Notes, which is helpful in talking about what this library + can do. . Certain parts of the API was inspired by pycassa (Python client) and hscassandra (on Hackage). @@ -24,11 +27,19 @@ implementation of Cassandra interaction using the thrift API underneath. . - * /Database.Cassandra.JSON/: A higher level API that operates on - values with ToJSON and FromJSON isntances from the /aeson/ - library. This module has in part been inspired by Bryan - O\'Sullivan\'s /riak/ client for Haskell. + * /Database.Cassandra.Marshall/: Intended to be the main high level + module that you should use, Marshall allows you to pick the + serialization strategy you would like to use at each function + call. We recommend using 'casSafeCopy' due to its support for + evolving data types, although casJSON maybe another popular + choice. . + * /Database.Cassandra.JSON/: (Now deprecated; use Marshall instead) + A higher level API that operates on values with ToJSON and + FromJSON isntances from the /aeson/ library. This module has in + part been inspired by Bryan O\'Sullivan\'s /riak/ client for + Haskell. + . * /Database.Cassandra.Pool/: Handles a /pool/ of connections to multiple servers in a cluster, splitting the load among them. . @@ -53,6 +64,7 @@ hs-source-dirs: src Exposed-modules: Database.Cassandra.Basic + Database.Cassandra.Marshall Database.Cassandra.JSON Database.Cassandra.Pool Database.Cassandra.Types @@ -62,6 +74,8 @@ base >= 4 && < 5 , bytestring , binary + , cereal + , safecopy , containers , network , time @@ -69,13 +83,17 @@ , stm , syb , text - , attoparsec >= 0.10 && < 0.11 + , attoparsec >= 0.10 && < 0.11 , aeson , Thrift >= 0.6 , cassandra-thrift >= 0.8 , resource-pool - , errors , data-default + , async + , errors + , MonadCatchIO-transformers >= 0.3 + , retry + , conduit >= 0.5 test-suite test @@ -87,6 +105,8 @@ base >= 4 && < 5 , bytestring , binary + , cereal + , safecopy , containers , network , time @@ -99,8 +119,10 @@ , Thrift >= 0.6 , cassandra-thrift >= 0.8 , resource-pool - , errors , data-default + , errors + , MonadCatchIO-transformers >= 0.3 + , retry , test-framework >= 0.6 , test-framework-quickcheck2 >= 0.2.12.2 diff --git a/src/Database/Cassandra/Basic.hs b/src/Database/Cassandra/Basic.hs --- a/src/Database/Cassandra/Basic.hs +++ b/src/Database/Cassandra/Basic.hs @@ -7,6 +7,18 @@ {-# LANGUAGE PatternGuards #-} {-# LANGUAGE RecordWildCards #-} +----------------------------------------------------------------------------- +-- | +-- Module : Database.Cassandra.Basic +-- Copyright : Ozgun Ataman +-- License : BSD3 +-- +-- Maintainer : Ozgun Ataman +-- Stability : experimental +-- +-- Low-level functionality for working with Cassandra at the most +-- basic level. +---------------------------------------------------------------------------- module Database.Cassandra.Basic @@ -24,6 +36,8 @@ , MonadCassandra (..) , Cas (..) , runCas + , transCas + , mapCassandra -- * Cassandra Operations , getCol @@ -32,9 +46,15 @@ , insert , delete + -- * Retrying Queries + , retryCas + , casRetryH + , networkRetryH + -- * Filtering , Selector(..) , range + , boundless , Order(..) , reverseOrder , KeySelector(..) @@ -81,15 +101,19 @@ ------------------------------------------------------------------------------- import Control.Applicative +import Control.Concurrent.Async import Control.Exception import Control.Monad +import qualified Control.Monad.CatchIO as MCIO import Control.Monad.Reader +import Control.Retry as R import Data.ByteString.Lazy (ByteString) import Data.Map (Map) import qualified Data.Map as M import Data.Maybe (mapMaybe) +import Data.Traversable (Traversable) import qualified Database.Cassandra.Thrift.Cassandra_Client as C -import Database.Cassandra.Thrift.Cassandra_Types (ConsistencyLevel(..)) +import Database.Cassandra.Thrift.Cassandra_Types (ConsistencyLevel (..)) import qualified Database.Cassandra.Thrift.Cassandra_Types as T import Prelude hiding (catch) ------------------------------------------------------------------------------- @@ -132,6 +156,15 @@ ------------------------------------------------------------------------------- +-- | Run a list of cassandra computations in parallel using the async library +mapCassandra :: (Traversable t, MonadCassandra m) => t (Cas b) -> m (t b) +mapCassandra ms = do + cp <- getCassandraPool + let f m = runCas cp m + liftIO $ mapConcurrently f ms + + +------------------------------------------------------------------------------- withCassandraPool :: MonadCassandra m => (Cassandra -> IO b) -> m b withCassandraPool f = do p <- getCassandraPool @@ -150,6 +183,19 @@ runCas = flip runReaderT +-- | Unwrap a Cassandra action and return an IO continuation that can +-- then be run in a pure IO context. +-- +-- This is useful when you design all your functions in a generic form +-- with 'MonadCassandra' m constraints and then one day need to feed +-- your function to a utility that can only run in an IO context. This +-- function is then your friendly utility for extracting an IO action. +transCas :: MonadCassandra m => Cas a -> m (IO a) +transCas m = do + cp <- getCassandraPool + return $ runCas cp m + + ------------------------------------------------------------------------------- instance (MonadIO m) => MonadCassandra (ReaderT CPool m) where getCassandraPool = ask @@ -252,7 +298,7 @@ insert cf k cl row = withCassandraPool $ \ Cassandra{..} -> do let insCol cp c = do c' <- mkThriftCol c - C.insert (cProto, cProto) k cp c' cl + wrapException $ C.insert (cProto, cProto) k cp c' cl forM_ row $ \ c -> do case c of Column{} -> do @@ -302,7 +348,7 @@ C.remove (cProto, cProto) k (cpCol c) now cl SupNames sn cs -> forM_ cs $ \c -> do C.remove (cProto, cProto) k (cpSCol sn c) now cl - Range _ _ _ _ -> error "delete: Range delete not implemented" + Range{} -> error "delete: Range delete not implemented" where -- wipe out the entire row cpAll = T.ColumnPath (Just cf) Nothing Nothing @@ -342,3 +388,17 @@ case res of Left e -> throw e Right a -> return a + + +-- | 'retrying' with direct cassandra support. Server-related failures +-- will be retried. +-- +-- 'UnavailableException', 'TimedOutException' and +-- 'SchemaDisagreementException' will be automatically retried. +retryCas :: MCIO.MonadCatchIO m + => R.RetrySettings + -- ^ For default settings, just use 'def' + -> m a + -- ^ Action to perform + -> m a +retryCas set f = R.recovering set [casRetryH, networkRetryH] f diff --git a/src/Database/Cassandra/JSON.hs b/src/Database/Cassandra/JSON.hs --- a/src/Database/Cassandra/JSON.hs +++ b/src/Database/Cassandra/JSON.hs @@ -36,6 +36,8 @@ , MonadCassandra (..) , Cas , runCas + , transCas + , mapCassandra -- * Cassandra Operations , get @@ -43,6 +45,7 @@ , getCol , getMulti , insertCol + , insertColTTL , modify , modify_ , delete @@ -58,6 +61,7 @@ -- * Filtering , Selector (..) , range + , boundless , Order(..) , reverseOrder , KeySelector (..) @@ -85,7 +89,7 @@ import Control.Monad import Data.Aeson as A import Data.Aeson.Parser (value) -import qualified Data.Attoparsec as Atto (IResult(..), parse) +import qualified Data.Attoparsec as Atto (IResult (..), parse) import qualified Data.ByteString.Char8 as B import Data.ByteString.Lazy.Char8 (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB @@ -94,7 +98,8 @@ import qualified Data.Map as M import Prelude hiding (catch) ------------------------------------------------------------------------------- -import Database.Cassandra.Basic hiding (KeySelector(..), delete, get, getCol, getMulti) +import Database.Cassandra.Basic hiding (KeySelector (..), delete, + get, getCol, getMulti) import qualified Database.Cassandra.Basic as CB ------------------------------------------------------------------------------- @@ -153,17 +158,16 @@ -- ^ Return the decided 'ModifyOperation' and its execution outcome modify cf k cn rcl wcl f = let - k' = toColKey k cn' = encodeCas cn execF prev = do (fres, b) <- f prev case fres of - Update a -> insert cf k' wcl [col cn' (marshallJSON' a)] - Delete -> CB.delete cf k' (ColNames [cn']) wcl + Update a -> insert cf k wcl [col cn' (marshallJSON' a)] + Delete -> CB.delete cf k (ColNames [cn']) wcl DoNothing -> return () return b in do - res <- CB.getCol cf k' cn' rcl + res <- CB.getCol cf k cn' rcl case res of Nothing -> execF Nothing Just Column{..} -> execF (unMarshallJSON' colVal) @@ -212,9 +216,29 @@ -> a -- ^ Content -> m () insertCol cf rk cn cl a = - insert cf (toColKey rk) cl [packCol (cn, marshallJSON' a)] + insert cf rk cl [packCol (cn, marshallJSON' a)] + +------------------------------------------------------------------------------- +-- Simple insertion function making use of typeclasses +insertColTTL + :: (MonadCassandra m, ToJSON a, CasType k) + => ColumnFamily + -> RowKey + -> k + -- ^ Column name. See 'CasType' for what you can use here. + -> ConsistencyLevel + -> a + -- ^ Content + -> Int32 + -- ^ TTL for this column + -> m () +insertColTTL cf rk cn cl a ttl = insert cf rk cl [column] + where + column = Column (packKey cn) (marshallJSON' a) Nothing (Just ttl) + + ------------------------------------------------------------------------------ -- | An arbitrary get operation - slice with 'Selector'. -- @@ -231,7 +255,7 @@ -> m [(k, a)] -- ^ List of key-value pairs. See 'CasType' for what key types you can use. get cf k s cl = do - res <- CB.get cf (toColKey k) s cl + res <- CB.get cf k s cl return $ map col2val res @@ -279,7 +303,7 @@ res <- CB.getMulti cf (ksToBasicKS ks) s cl return . M.fromList . map conv . M.toList $ res where - conv (k, row) = (fromColKey' k, map col2val row) + conv (k, row) = (k, map col2val row) ------------------------------------------------------------------------------- @@ -293,7 +317,7 @@ -> ConsistencyLevel -> m (Maybe a) getCol cf rk ck cl = do - res <- CB.getCol cf (toColKey rk) (encodeCas ck) cl + res <- CB.getCol cf rk (encodeCas ck) cl case res of Nothing -> return Nothing Just res' -> do @@ -315,7 +339,7 @@ -- ^ Columns to be deleted -> ConsistencyLevel -> m () -delete cf k s cl = CB.delete cf (toColKey k) s cl +delete cf k s cl = CB.delete cf k s cl ------------------------------------------------------------------------------ @@ -334,6 +358,7 @@ -- | Lazy 'unMarshallJSON' unMarshallJSON' :: FromJSON a => ByteString -> Maybe a unMarshallJSON' = unMarshallJSON . B.concat . LB.toChunks + ------------------------------------------------------------------------------ -- | Decode JSON diff --git a/src/Database/Cassandra/Marshall.hs b/src/Database/Cassandra/Marshall.hs new file mode 100644 --- /dev/null +++ b/src/Database/Cassandra/Marshall.hs @@ -0,0 +1,470 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeSynonymInstances #-} + + +----------------------------------------------------------------------------- +-- | +-- Module : +-- Copyright : (C) 2013 Ozgun Ataman +-- License : All Rights Reserved +-- +-- Maintainer : Ozgun Ataman +-- Stability : experimental +-- +-- Defines Cassandra operations for persistence of complex Haskell +-- data objects with custom-selected but implicitly performed +-- serialization. +-- +-- The main design choice is to require a dictionary dictating +-- marshalling/serialization policy for every operation, rather than a +-- typeclass that can be instantiated once. +---------------------------------------------------------------------------- + +module Database.Cassandra.Marshall + ( + + -- * Connection + CPool + , Server + , defServer + , defServers + , KeySpace + , createCassandraPool + + -- * MonadCassandra Typeclass + , MonadCassandra (..) + , Cas + , runCas + , transCas + , mapCassandra + + -- * Haskell Record Marshalling + + , Marshall (..) + , casShow + , casJSON + -- , casBinary + , casSerialize + , casSafeCopy + + -- * Cassandra Operations + , get + , get_ + , getCol + , getMulti + , insertCol + , insertColTTL + , modify + , modify_ + , delete + + -- * Retrying Queries + , CB.retryCas + , casRetryH + + -- * Necessary Types + , RowKey + , ColumnName + , ModifyOperation (..) + , ColumnFamily + , ConsistencyLevel (..) + , CassandraException (..) + + -- * Filtering + , Selector (..) + , range + , boundless + , Order(..) + , reverseOrder + , KeySelector (..) + , KeyRangeType (..) + + + -- * Pagination + , PageResult (..) + , pIsDry + , pIsDone + , pHasMore + , paginate + , paginateSource + , pageToSource + + -- * Helpers + , CKey (..) + , fromColKey' + + -- * Cassandra Column Key Types + , module Database.Cassandra.Pack + ) where + +------------------------------------------------------------------------------- +import Control.Error +import Control.Monad +import Control.Monad.CatchIO +import Control.Monad.Trans +import Control.Retry as R +import qualified Data.Aeson as A +import qualified Data.Attoparsec as Atto (IResult (..), parse) +import qualified Data.Binary as BN +import qualified Data.Binary.Get as BN +import qualified Data.ByteString.Char8 as B +import Data.ByteString.Lazy.Char8 (ByteString) +import qualified Data.ByteString.Lazy.Char8 as LB +import Data.Conduit +import qualified Data.Conduit.List as C +import Data.Int (Int32) +import Data.Map (Map) +import qualified Data.Map as M +import qualified Data.SafeCopy as SC +import qualified Data.Serialize as SL +import Prelude hiding (catch) +------------------------------------------------------------------------------- +import Database.Cassandra.Basic hiding (KeySelector (..), delete, + get, getCol, getMulti) +import qualified Database.Cassandra.Basic as CB +import Database.Cassandra.Pack +import Database.Cassandra.Types +------------------------------------------------------------------------------- + + +-- | A Haskell dictionary containing a pair of encode/decode +-- functions. +-- +-- This is the main design choice in this module. We require that each +-- operation takes an explicit marshalling policy rather than a +-- typeclass which makes it possible to do it in a single way per data +-- type. +-- +-- You can create your own objects of this type with great ease. Just +-- look at one of the examples here ('casJSON', 'casSerialize', etc.) +data Marshall a = Marshall { + marshallEncode :: a -> ByteString + -- ^ An encoding function + , marshallDecode :: ByteString -> Either String a + -- ^ A decoding function + } + + +-- | Marshall data using JSON encoding. Good interoperability, but not +-- very efficient for data storage. +casJSON :: (A.ToJSON a, A.FromJSON a) => Marshall a +casJSON = Marshall A.encode A.eitherDecode + + +-- | Marshall data using 'Show' and 'Read'. Not meant for serious production cases. +casShow :: (Show a, Read a) => Marshall a +casShow = Marshall + (LB.pack . show) + (readErr "casShow can't read cassandra value" . LB.unpack) + + +-- -- | Marshall data using the 'Binary' instance. This is one of the +-- -- efficient methods available. +-- casBinary :: BN.Binary a => Marshall a +-- casBinary = Marshall BN.encode dec +-- where +-- dec bs = case BN.runGetOrFail BN.get bs of +-- Left (_,_,err) -> Left err +-- Right (_,_,a) -> Right a + + +-- | Marshall data using the 'SafeCopy' instance. This is quite well +-- suited for production because it is both very efficient and +-- provides a systematic way to migrate your data types over time. +casSafeCopy :: SC.SafeCopy a => Marshall a +casSafeCopy = Marshall (SL.runPutLazy . SC.safePut) (SL.runGetLazy SC.safeGet) + + +-- | Marshall data using the 'Serialize' instance. Like 'Binary', +-- 'Serialize' is very efficient. +casSerialize :: SL.Serialize a => Marshall a +casSerialize = Marshall SL.encodeLazy SL.decodeLazy + + +------------------------------------------------------------------------------ +-- | A modify function that will fetch a specific column, apply modification +-- function on it and save results back to Cassandra. +-- +-- A 'b' side value is returned for computational convenience. +-- +-- This is intended to be a workhorse function, in that you should be +-- able to do all kinds of relatively straightforward operations just +-- using this function. +-- +-- This method may throw a 'CassandraException' for all exceptions other than +-- 'NotFoundException'. +modify + :: (MonadCassandra m, CasType k) + => Marshall a + -- ^ A serialization methodology. Example: 'casJSON' + -> ColumnFamily + -> RowKey + -> k + -- ^ Column name; anything in CasType + -> ConsistencyLevel + -- ^ Read quorum + -> ConsistencyLevel + -- ^ Write quorum + -> (Maybe a -> m (ModifyOperation a, b)) + -- ^ Modification function. Called with 'Just' the value if present, + -- 'Nothing' otherwise. + -> m b + -- ^ Return the decided 'ModifyOperation' and its execution outcome +modify Marshall{..} cf k cn rcl wcl f = + let + cn' = encodeCas cn + execF prev = do + (fres, b) <- f prev + case fres of + Update a -> insert cf k wcl [col cn' (marshallEncode a)] + Delete -> CB.delete cf k (ColNames [cn']) wcl + DoNothing -> return () + return b + in do + res <- CB.getCol cf k cn' rcl + case res of + Nothing -> execF Nothing + Just Column{..} -> execF (hush $ marshallDecode colVal) + Just SuperColumn{..} -> throw $ + OperationNotSupported "modify not implemented for SuperColumn" + + +------------------------------------------------------------------------------ +-- | Same as 'modify' but does not offer a side value. +-- +-- This method may throw a 'CassandraException' for all exceptions other than +-- 'NotFoundException'. +modify_ + :: (MonadCassandra m, CasType k) + => Marshall a + -> ColumnFamily + -> RowKey + -> k + -- ^ Column name; anything in CasType + -> ConsistencyLevel + -- ^ Read quorum + -> ConsistencyLevel + -- ^ Write quorum + -> (Maybe a -> m (ModifyOperation a)) + -- ^ Modification function. Called with 'Just' the value if present, + -- 'Nothing' otherwise. + -> m () +modify_ m cf k cn rcl wcl f = + let + f' prev = do + op <- f prev + return (op, ()) + in do + modify m cf k cn rcl wcl f' + return () + + +------------------------------------------------------------------------------- +-- Simple insertion function making use of typeclasses +insertCol + :: (MonadCassandra m, CasType k) + => Marshall a + -> ColumnFamily + -> RowKey + -> k + -- ^ Column name. See 'CasType' for what you can use here. + -> ConsistencyLevel + -> a -- ^ Content + -> m () +insertCol Marshall{..} cf rk cn cl a = + insert cf rk cl [packCol (cn, marshallEncode a)] + + + +------------------------------------------------------------------------------- +-- Simple insertion function making use of typeclasses +insertColTTL + :: (MonadCassandra m, CasType k) + => Marshall a + -> ColumnFamily + -> RowKey + -> k + -- ^ Column name. See 'CasType' for what you can use here. + -> ConsistencyLevel + -> a + -- ^ Content + -> Int32 + -- ^ TTL for this column + -> m () +insertColTTL Marshall{..} cf rk cn cl a ttl = insert cf rk cl [column] + where + column = Column (packKey cn) (marshallEncode a) Nothing (Just ttl) + + +------------------------------------------------------------------------------ +-- | An arbitrary get operation - slice with 'Selector'. +-- +-- Internally based on Basic.get. Table is assumed to be a regular +-- ColumnFamily and contents of returned columns are cast into the +-- target type. +get + :: (MonadCassandra m, CasType k) + => Marshall a + -> ColumnFamily + -> RowKey + -> Selector + -- ^ A slice selector + -> ConsistencyLevel + -> m [(k, a)] + -- ^ List of key-value pairs. See 'CasType' for what key types you can use. +get m cf k s cl = do + res <- CB.get cf k s cl + return $ map (col2val m) res + + +------------------------------------------------------------------------------- +-- | A version of 'get' that discards the column names for the common +-- scenario. Useful because you would otherwise be forced to manually +-- supply type signatures to get rid of the 'CasType' ambiguity. +get_ + :: (MonadCassandra m) + => Marshall a + -> ColumnFamily + -> RowKey + -> Selector + -- ^ A slice selector + -> ConsistencyLevel + -> m [a] +get_ m cf k s cl = do + (res :: [(LB.ByteString, a)]) <- get m cf k s cl + return $ map snd res + + +------------------------------------------------------------------------------- +ksToBasicKS :: KeySelector -> CB.KeySelector +ksToBasicKS (Keys k) = CB.Keys $ map toColKey k +ksToBasicKS (KeyRange ty fr to i) = CB.KeyRange ty (toColKey fr) (toColKey to) i + + +------------------------------------------------------------------------------- +-- | Get a slice of columns from multiple rows at once. Note that +-- since we are auto-serializing from JSON, all the columns must be of +-- the same data type. +getMulti + :: (MonadCassandra m) + => Marshall a + -> ColumnFamily + -> KeySelector + -> Selector + -> ConsistencyLevel + -> m (Map RowKey [(ColumnName, a)]) +getMulti m cf ks s cl = do + res <- CB.getMulti cf (ksToBasicKS ks) s cl + return . M.fromList . map conv . M.toList $ res + where + conv (k, row) = (k, map (col2val m) row) + + +------------------------------------------------------------------------------- +-- | Get a single column from a single row +getCol + :: (MonadCassandra m, CasType k) + => Marshall a + -> ColumnFamily + -> RowKey + -> k + -- ^ Column name; anything in 'CasType' + -> ConsistencyLevel + -> m (Maybe a) +getCol m cf rk ck cl = do + res <- CB.getCol cf rk (encodeCas ck) cl + case res of + Nothing -> return Nothing + Just res' -> do + let (_ :: ByteString, x) = col2val m res' + return $ Just x + + +------------------------------------------------------------------------------ +-- | Same as the 'delete' in the 'Cassandra.Basic' module, except that +-- it throws an exception rather than returning an explicit Either +-- value. +delete + :: (MonadCassandra m) + => ColumnFamily + -- ^ In 'ColumnFamily' + -> RowKey + -- ^ Key to be deleted + -> Selector + -- ^ Columns to be deleted + -> ConsistencyLevel + -> m () +delete cf k s cl = CB.delete cf k s cl + + +------------------------------------------------------------------------------- +-- | Convert regular column to a key-value pair +col2val :: CasType k => Marshall a -> Column -> (k, a) +col2val Marshall{..} c = f $ unpackCol c + where + f (k, val) = (k, either err id $ marshallDecode val) + err s = error $ "Cassandra Marshall: Value can't be decoded: " ++ s +col2val _ _ = error "col2val is not implemented for SuperColumns" + + + +------------------------------------------------------------------------------- +-- | Paginate over columns in a given key, repeatedly applying the +-- given 'Selector'. The 'Selector' must be a 'Range' selector, or +-- else this funtion will raise an exception. +paginate + :: (MonadCassandra m, MonadCatchIO m, CasType k) + => Marshall a + -- ^ Serialization strategy + -> ColumnFamily + -> RowKey + -- ^ Paginate columns of this row + -> Selector + -- ^ 'Range' selector to initially and repeatedly apply. + -> ConsistencyLevel + -> RetrySettings + -- ^ Retry strategy for each underlying Cassandra call + -> m (PageResult m (k, a)) +paginate m cf k rng@(Range from to ord per) cl retry = do + cs <- reverse `liftM` retryCas retry (get m cf k rng cl) + case cs of + [] -> return $ PDone [] + [a] -> return $ PDone [a] + _ -> + let cont = paginate m cf k (Range (Just cn) to ord per) cl retry + (cn, _) = head cs + in return $ PMore (reverse (drop 1 cs)) cont +paginate _ _ _ _ _ _ = error "Must call paginate with a Range selector" + + + +------------------------------------------------------------------------------- +-- | Convenience layer: Convert a pagination scheme to a conduit 'Source'. +pageToSource :: (MonadCatchIO m, Monad m) => PageResult m a -> Source m a +pageToSource (PDone as) = C.sourceList as +pageToSource (PMore as m) = C.sourceList as >> lift m >>= pageToSource + + +------------------------------------------------------------------------------- +-- | Just like 'paginate', but we instead return a conduit 'Source'. +paginateSource + :: (CasType k, MonadCatchIO m, MonadCassandra m) + => Marshall a + -> ColumnFamily + -> RowKey + -> Selector + -> ConsistencyLevel + -> RetrySettings + -> Source m (k, a) +paginateSource m cf k rng cl r = do + buf <- lift $ paginate m cf k rng cl r + pageToSource buf + diff --git a/src/Database/Cassandra/Pack.hs b/src/Database/Cassandra/Pack.hs --- a/src/Database/Cassandra/Pack.hs +++ b/src/Database/Cassandra/Pack.hs @@ -1,6 +1,7 @@ -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE OverlappingInstances #-} -{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverlappingInstances #-} +{-# LANGUAGE RankNTypes #-} {-| A Collection of utilities for binary packing values into Bytestring |-} @@ -14,8 +15,13 @@ , TUtf8 (..) , TUUID (..) , TLong (..) + , TTimeStamp (..) + , toTimeStamp + , fromTimeStamp + , Exclusive (..) , Single (..) + , SliceStart (..) ) where ------------------------------------------------------------------------------- @@ -33,6 +39,9 @@ import qualified Data.Text.Encoding as T import qualified Data.Text.Lazy as LT import qualified Data.Text.Lazy.Encoding as LT +import Data.Time +import Data.Time +import Data.Time.Clock.POSIX ------------------------------------------------------------------------------- @@ -42,12 +51,32 @@ newtype TBytes = TBytes { getTBytes :: ByteString } deriving (Eq,Show,Read,Ord) newtype TCounter = TCounter { getCounter :: ByteString } deriving (Eq,Show,Read,Ord) newtype TInt32 = TInt32 { getInt32 :: Int32 } deriving (Eq,Show,Read,Ord) -newtype TInt = TInt { getInt :: Integer } deriving (Eq,Show,Read,Ord) +newtype TInt = TInt { getInt :: Integer } deriving (Eq,Show,Read,Ord,Enum,Real,Integral,Num) newtype TUUID = TUUID { getUUID :: ByteString } deriving (Eq,Show,Read,Ord) -newtype TLong = TLong { getLong :: Integer } deriving (Eq,Show,Read,Ord) +newtype TLong = TLong { getLong :: Integer } deriving (Eq,Show,Read,Ord,Enum,Real,Integral,Num) newtype TUtf8 = TUtf8 { getUtf8 :: Text } deriving (Eq,Show,Read,Ord) +-- | Timestamp that stores micro-seconds since epoch as 'TLong' underneath. +newtype TTimeStamp = TTimeStamp { getTimeStamp :: TLong } + deriving (Eq,Show,Read,Ord,Enum,Num,Real,Integral,CasType) + + +-- | Convert commonly used 'UTCTime' to 'TTimeStamp'. +-- +-- First converts to seconds since epoch (POSIX seconds), then +-- multiplies by a million and floors the resulting value. The value, +-- therefore, is in micro-seconds and is accurate to within a +-- microsecond. +toTimeStamp :: UTCTime -> TTimeStamp +toTimeStamp utc = fromIntegral . floor . (* 1e6) $ utcTimeToPOSIXSeconds utc + + +fromTimeStamp :: TTimeStamp -> UTCTime +fromTimeStamp (TTimeStamp (TLong i)) = + posixSecondsToUTCTime $ realToFrac $ fromIntegral i / (1e6) + + ------------------------------------------------------------------------------- -- | This typeclass defines and maps to haskell types that Cassandra -- natively knows about and uses in sorting and potentially validating @@ -131,12 +160,16 @@ encodeCas = runPut . putWord64be . fromIntegral . getInt decodeCas = TInt . fromIntegral . runGet getWord64be + +------------------------------------------------------------------------------- instance CasType Int where encodeCas = encodeCas . TInt . fromIntegral decodeCas = fromIntegral . getInt . decodeCas + ------------------------------------------------------------------------------- --- | Pack as an 8 byte unsigned number; negative signs are lost. +-- | Pack as an 8 byte unsigned number; negative signs are lost. Maps +-- to 'LongType'. instance CasType TLong where encodeCas = runPut . putWord64be . fromIntegral . getLong decodeCas = TLong . fromIntegral . runGet getWord64be @@ -150,14 +183,27 @@ ------------------------------------------------------------------------------- +-- | Encode days as 'LongType' via 'TLong'. +instance CasType Day where + encodeCas = encodeCas . TLong . toModifiedJulianDay + decodeCas = ModifiedJulianDay . getLong . decodeCas + + +-- | Via 'TTimeStamp', which is via 'TLong' +instance CasType UTCTime where + encodeCas = encodeCas . toTimeStamp + decodeCas = fromTimeStamp . decodeCas + + +------------------------------------------------------------------------------- -- | Use the 'Single' wrapper when querying only with the first of a -- two or more field CompositeType. instance (CasType a) => CasType (Single a) where - encodeCas a = runPut $ do - putSegment a end + encodeCas (Single a) = runPut $ putSegment a end decodeCas bs = flip runGet bs $ Single <$> getSegment + ------------------------------------------------------------------------------- -- | Composite types - see Cassandra or pycassa docs to understand instance (CasType a, CasType b) => CasType (a,b) where @@ -196,6 +242,77 @@ <*> getSegment + ------------------ + -- Slice Starts -- + ------------------ + + + +instance (CasType a) => CasType (SliceStart (Single a)) where + encodeCas (SliceStart (Single a)) = runPut $ do + putSegment a exc + decodeCas bs = flip runGet bs $ (SliceStart . Single) <$> getSegment + + +------------------------------------------------------------------------------- +-- | Composite types - see Cassandra or pycassa docs to understand +instance (CasType a, CasType b) => CasType (SliceStart (a,b)) where + encodeCas (SliceStart (a, b)) = runPut $ do + putSegment a sep + putSegment b exc + + decodeCas bs = SliceStart . flip runGet bs $ (,) + <$> getSegment + <*> getSegment + + +instance (CasType a, CasType b, CasType c) => CasType (SliceStart (a,b,c)) where + encodeCas (SliceStart (a, b, c)) = runPut $ do + putSegment a sep + putSegment b sep + putSegment c exc + + decodeCas bs = SliceStart . flip runGet bs $ (,,) + <$> getSegment + <*> getSegment + <*> getSegment + + +instance (CasType a, CasType b, CasType c, CasType d) => + CasType (SliceStart (a,b,c,d)) where + encodeCas (SliceStart (a, b, c, d)) = runPut $ do + putSegment a sep + putSegment b sep + putSegment c sep + putSegment d exc + + decodeCas bs = SliceStart . flip runGet bs $ (,,,) + <$> getSegment + <*> getSegment + <*> getSegment + <*> getSegment + + + + + + + + + + + ----------------------- + -- Exclusive Columns -- + ----------------------- + + +instance CasType a => CasType (Exclusive (Single a)) where + encodeCas (Exclusive (Single a)) = runPut $ do + putSegment a exc + + decodeCas = Exclusive . decodeCas + + instance (CasType a, CasType b) => CasType (a, Exclusive b) where encodeCas (a, Exclusive b) = runPut $ do putSegment a sep @@ -249,6 +366,12 @@ -- | Use the Single wrapper when you want to refer only to the first -- coolumn of a CompositeType column. newtype Single a = Single a deriving (Eq,Show,Read,Ord) + + +------------------------------------------------------------------------------- +-- | Wrap your composite columns in this type when you're starting an +-- inclusive column slice. +newtype SliceStart a = SliceStart a deriving (Eq,Show,Read,Ord) -- | composite columns are a pain diff --git a/src/Database/Cassandra/Pool.hs b/src/Database/Cassandra/Pool.hs --- a/src/Database/Cassandra/Pool.hs +++ b/src/Database/Cassandra/Pool.hs @@ -95,27 +95,36 @@ -> IO CPool createCassandraPool servers numStripes perStripe maxIdle ks = do sring <- newTVarIO $ mkRing servers - pool <- createPool (cr sring) dest numStripes maxIdle perStripe - forkIO (serverDiscoveryThread sring ks pool) + pool <- createPool (cr 4 sring) dest numStripes maxIdle perStripe + -- forkIO (serverDiscoveryThread sring ks pool) return pool where - cr :: ServerRing -> IO Cassandra - cr sring = do + cr :: Int -> ServerRing -> IO Cassandra + cr n sring = do s@(host, p) <- atomically $ do ring@Ring{..} <- readTVar sring writeTVar sring (next ring) return current - handle (handler sring s) $ do + handle (handler n sring s) $ do (h,ft,proto) <- openThrift host p C.set_keyspace (proto, proto) ks return $ Cassandra h ft proto - handler :: ServerRing -> Server -> SomeException -> IO Cassandra - handler sring server e = do - modifyServers sring (removeServer server) - cr sring + handler :: Int -> ServerRing -> Server -> SomeException -> IO Cassandra + handler 0 _ _ e = error $ "Can't connect to cassandra after several tries: " ++ show e + handler n sring server e = do + -- we need a temporary removal system for servers; something + -- with a TTL just removing them from ring is dangerous, what if + -- the network is partitioned for a little while? + + -- modifyServers sring (removeServer server) + + -- wait 100ms to avoid crazy loops + threadDelay 100000 + cr (n-1) sring + dest h = hClose $ cHandle h @@ -171,7 +180,7 @@ ------------------------------------------------------------------------------ mkRing [] = error "Can't make a ring from empty list" -mkRing (a:as) = Ring S.empty a [] as +mkRing all@(a:as) = Ring (S.fromList all) a [] as ------------------------------------------------------------------------------ @@ -187,11 +196,11 @@ ------------------------------------------------------------------------------ removeServer :: Ord a => a -> Ring a -> Ring a removeServer s r@Ring{..} - | s `S.member` allItems = Ring (S.delete s allItems) current' used' upcoming' + | s `S.member` allItems = Ring all' cur' [] up' | otherwise = r where - used' = filter (/=s) used - (current':upcoming') = filter (/=s) (current:upcoming) + all' = S.delete s allItems + cur' : up' = S.toList all' ------------------------------------------------------------------------------ diff --git a/src/Database/Cassandra/Types.hs b/src/Database/Cassandra/Types.hs --- a/src/Database/Cassandra/Types.hs +++ b/src/Database/Cassandra/Types.hs @@ -5,14 +5,15 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternGuards #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeSynonymInstances #-} module Database.Cassandra.Types where ------------------------------------------------------------------------------- -import Control.Error import Control.Exception import Control.Monad +import qualified Control.Monad.CatchIO as MCIO import qualified Data.ByteString.Char8 as B import Data.ByteString.Lazy (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB @@ -33,6 +34,15 @@ ------------------------------------------------------------------------------- +------------------------------------------------------------------------------ +-- | Possible outcomes of a modify operation +data ModifyOperation a = + Update a + | Delete + | DoNothing + deriving (Eq,Show,Ord,Read) + + -- | A 'Key' range selector to use with 'getMulti'. data KeySelector = Keys [Key] @@ -73,7 +83,7 @@ -- ^ When deleting specific columns in a super column | forall a b. (CasType a, CasType b) => Range { rangeStart :: Maybe a - , rangeEnd :: Maybe b + , rangeEnd :: Maybe b , rangeOrder :: Order , rangeLimit :: Int32 } @@ -87,7 +97,11 @@ range = Range (Nothing :: Maybe ByteString) (Nothing :: Maybe ByteString) Regular 1024 +boundless :: Maybe ByteString +boundless = Nothing + + instance Default Selector where def = All @@ -111,7 +125,7 @@ mkPredicate :: Selector -> C.SlicePredicate mkPredicate s = let - allRange = C.SliceRange (Just "") (Just "") (Just False) (Just 5000) + allRange = C.SliceRange (Just "") (Just "") (Just False) (Just 50000) in case s of All -> C.SlicePredicate Nothing (Just allRange) ColNames ks -> C.SlicePredicate (Just (map encodeCas ks)) Nothing @@ -144,6 +158,7 @@ type Key = ByteString +type RowKey = Key type ColumnName = ByteString @@ -160,7 +175,7 @@ | Column { colKey :: ColumnName , colVal :: Value - , colTS :: Maybe Int64 + , colTS :: Maybe Int64 -- ^ Last update timestamp; will be overridden during write/update ops , colTTL :: Maybe Int32 -- ^ A TTL after which Cassandra will erase the column @@ -229,6 +244,23 @@ instance Exception CassandraException +-- | Exception handler that returns @True@ for errors that may be +-- resolved after a retry. So they are good candidates for 'retrying' +-- queries. +casRetryH :: Monad m => MCIO.Handler m Bool +casRetryH = MCIO.Handler $ \ e -> return $ + case e of + UnavailableException{} -> True + TimedOutException{} -> True + SchemaDisagreementException{} -> True + _ -> False + + +-- | 'IOException's should be retried +networkRetryH :: Monad m => MCIO.Handler m Bool +networkRetryH = MCIO.Handler $ \ (_ :: IOException) -> return True + + ------------------------------------------------------------------------------ -- | Cassandra is VERY sensitive to its timestamp values. As a convention, -- timestamps are always in microseconds @@ -236,6 +268,43 @@ getTime = do t <- getPOSIXTime return . fromIntegral . floor $ t * 1000000 + + + ---------------- + -- Pagination -- + ---------------- + + +------------------------------------------------------------------------------- +-- | Describes the result of a single pagination action +data PageResult m a + = PDone { pCache :: [a] } + -- ^ Done, this is all I have. + | PMore { pCache :: [a], pMore :: m (PageResult m a) } + -- ^ Here's a batch and there is more when you call the action. + + + + +------------------------------------------------------------------------------- +pIsDry x = pIsDone x && null (pCache x) + +------------------------------------------------------------------------------- +pIsDone PDone{} = True +pIsDone _ = False + + +------------------------------------------------------------------------------- +pHasMore PMore{} = True +pHasMore _ = False + + +------------------------------------------------------------------------------- +instance Monad m => Functor (PageResult m) where + fmap f (PDone as) = PDone (fmap f as) + fmap f (PMore as m) = PMore (fmap f as) m' + where + m' = liftM (fmap f) m diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -1,6 +1,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TemplateHaskell #-} @@ -15,10 +16,12 @@ import qualified Data.Map as M import qualified Data.Text as T import qualified Database.Cassandra.Thrift.Cassandra_Client as C -import Database.Cassandra.Thrift.Cassandra_Types (ConsistencyLevel(..)) +import Database.Cassandra.Thrift.Cassandra_Types + (ConsistencyLevel (..)) import Database.Cassandra.Thrift.Cassandra_Types as T import System.IO.Unsafe -import Test.Framework (defaultMain, testGroup) +import Test.Framework (defaultMain, + testGroup) import Test.Framework.Providers.HUnit import Test.Framework.Providers.QuickCheck2 (testProperty) import Test.HUnit @@ -47,7 +50,8 @@ , testProperty "cas type marshalling ascii" prop_casTypeInt32 , testProperty "cas type marshalling composite" prop_casTypeComp , testCase "cas live test - composite get/set" test_composite_col - , testProperty "cas live test read/write QC" (prop_composite_col_readWrite pool) + , testCase "cas live - set comp + single col slice" test_composite_slice + -- , testProperty "cas live test read/write QC" (prop_composite_col_readWrite pool) ] @@ -126,6 +130,26 @@ ------------------------------------------------------------------------------- +test_composite_slice = do + pool <- mkTestConn + xs <- runCas pool $ do + insert "testing" "row2" ONE [packCol (key, content)] + get "testing" "row2" slice ONE + let (res :: [((TLong, TBytes), LB.ByteString)]) = map unpackCol xs + + assertEqual "composite single col slice" content (snd . head $ res) + where + key = (TLong 125, TBytes "oklahoma") + content = "asdf" + slice = Range (Just (Exclusive (Single (TLong 125)))) + (Just (Single (TLong 125))) + -- (Just (Single (TLong 127))) + Regular + 100 + -- slice = Range (Just (TLong 125, TBytes "")) (Just (TLong 125, TBytes "zzzzz")) Regular 100 + + +------------------------------------------------------------------------------- -- | test quick-check generated pairs for composite column prop_composite_col_readWrite :: CPool -> ((TLong, TBytes), LB.ByteString) -> Property prop_composite_col_readWrite pool content@(k@(TLong i, _),v) = i >= 0 ==> @@ -134,6 +158,7 @@ insert "testing" "row" ONE [packCol content] getCol "testing" "row" (packKey k) ONE return $ (Just content) == fmap unpackCol res +