2010-11-01 00:38:38 +00:00
-- | Query and update documents
2010-06-15 03:14:40 +00:00
2010-12-20 02:08:53 +00:00
{- # LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, RankNTypes # -}
2010-06-15 03:14:40 +00:00
module Database.MongoDB.Query (
2010-10-27 20:13:23 +00:00
-- * Access
access , Access , Action , runAction , Failure ( .. ) ,
2010-06-15 03:14:40 +00:00
-- * Database
2010-10-27 20:13:23 +00:00
Database ( .. ) , allDatabases , DbAccess , use , thisDatabase ,
2010-06-15 03:14:40 +00:00
-- ** Authentication
P . Username , P . Password , auth ,
-- * Collection
Collection , allCollections ,
-- ** Selection
2010-06-21 15:06:20 +00:00
Selection ( .. ) , Selector , whereJS ,
Select ( select ) ,
2010-06-15 03:14:40 +00:00
-- * Write
2010-06-21 15:06:20 +00:00
-- ** WriteMode
2010-10-27 20:13:23 +00:00
WriteMode ( .. ) , safe , GetLastError , writeMode ,
2010-06-15 03:14:40 +00:00
-- ** Insert
insert , insert_ , insertMany , insertMany_ ,
-- ** Update
save , replace , repsert , Modifier , modify ,
-- ** Delete
delete , deleteOne ,
-- * Read
2010-11-01 00:38:38 +00:00
readMode ,
2010-06-15 03:14:40 +00:00
-- ** Query
2010-07-27 21:18:53 +00:00
Query ( .. ) , QueryOption ( .. ) , Projector , Limit , Order , BatchSize ,
2010-06-15 03:14:40 +00:00
explain , find , findOne , count , distinct ,
-- *** Cursor
2010-12-20 02:08:53 +00:00
Cursor , next , nextN , rest , closeCursor , isCursorClosed ,
2010-06-15 03:14:40 +00:00
-- ** Group
Group ( .. ) , GroupKey ( .. ) , group ,
-- ** MapReduce
MapReduce ( .. ) , MapFun , ReduceFun , FinalizeFun , mapReduce , runMR , runMR' ,
-- * Command
Command , runCommand , runCommand1 ,
eval ,
) where
import Prelude as X hiding ( lookup )
2010-06-21 15:06:20 +00:00
import Control.Applicative ( ( <$> ) , Applicative ( .. ) )
2010-07-27 21:18:53 +00:00
import Control.Arrow ( first )
2010-06-21 15:06:20 +00:00
import Control.Monad.Context
import Control.Monad.Reader
import Control.Monad.Error
2010-07-27 21:18:53 +00:00
import Control.Monad.Throw
2010-11-01 19:35:13 +00:00
import Control.Monad.MVar
2010-06-15 03:14:40 +00:00
import qualified Database.MongoDB.Internal.Protocol as P
2010-07-27 21:18:53 +00:00
import Database.MongoDB.Internal.Protocol hiding ( Query , QueryOption ( .. ) , send , call )
2010-12-27 05:23:02 +00:00
import Database.MongoDB.Connection ( MasterOrSlaveOk ( .. ) , Service ( .. ) )
2010-06-15 03:14:40 +00:00
import Data.Bson
import Data.Word
import Data.Int
2010-07-27 21:18:53 +00:00
import Data.Maybe ( listToMaybe , catMaybes )
2010-12-27 05:23:02 +00:00
import Data.UString as U ( dropWhile , any , tail )
2010-11-01 19:35:13 +00:00
import Control.Monad.Util ( MonadIO ' , loop )
2010-10-27 20:13:23 +00:00
import Database.MongoDB.Internal.Util ( ( <.> ) , true1 )
2010-06-21 15:06:20 +00:00
2010-10-27 20:13:23 +00:00
mapErrorIO :: ( Throw e m , MonadIO m ) => ( e' -> e ) -> ErrorT e' IO a -> m a
mapErrorIO f = throwLeft' f . liftIO . runErrorT
2010-07-03 17:15:30 +00:00
2010-10-27 20:13:23 +00:00
-- * Mongo Monad
2010-07-27 21:18:53 +00:00
2010-12-27 05:23:02 +00:00
access :: ( Service s , MonadIO m ) => WriteMode -> MasterOrSlaveOk -> ConnPool s -> Action m a -> m ( Either Failure a )
2010-11-01 00:38:38 +00:00
-- ^ Run action under given write and read mode against the server or replicaSet behind given connection pool. Return Left Failure if there is a connection failure or read/write error.
access w mos pool act = do
ePipe <- liftIO . runErrorT $ getPipe mos pool
2010-10-27 20:13:23 +00:00
either ( return . Left . ConnectionFailure ) ( runAction act w mos ) ePipe
2010-07-27 21:18:53 +00:00
2010-10-27 20:13:23 +00:00
-- | A monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws 'Failure' on read, write, or pipe failure
2010-11-01 19:35:13 +00:00
class ( Context Pipe m , Context MasterOrSlaveOk m , Context WriteMode m , Throw Failure m , MonadIO' m , MonadMVar m ) => Access m
instance ( Context Pipe m , Context MasterOrSlaveOk m , Context WriteMode m , Throw Failure m , MonadIO' m , MonadMVar m ) => Access m
2010-07-03 17:15:30 +00:00
2010-10-27 20:13:23 +00:00
newtype Action m a = Action ( ErrorT Failure ( ReaderT WriteMode ( ReaderT MasterOrSlaveOk ( ReaderT Pipe m ) ) ) a )
2010-11-01 19:35:13 +00:00
deriving ( Context Pipe , Context MasterOrSlaveOk , Context WriteMode , Throw Failure , MonadIO , MonadMVar , Monad , Applicative , Functor )
2010-10-27 20:13:23 +00:00
-- ^ Monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws a 'Failure' on read, write or pipe failure
2010-07-03 17:15:30 +00:00
2010-10-27 20:13:23 +00:00
instance MonadTrans Action where
lift = Action . lift . lift . lift . lift
2010-06-21 15:06:20 +00:00
2010-10-27 20:13:23 +00:00
runAction :: Action m a -> WriteMode -> MasterOrSlaveOk -> Pipe -> m ( Either Failure a )
2010-11-01 00:38:38 +00:00
-- ^ Run action with given write mode and read mode (master or slave-ok) against given pipe (TCP connection). Return Left Failure if read/write error or connection failure.
-- 'access' calls runAction. Use this directly if you want to use the same connection and not take from the pool again. However, the connection may still be used by other threads at the same time. For instance, the pool will still hand this connection out.
2010-10-27 20:13:23 +00:00
runAction ( Action action ) w mos = runReaderT ( runReaderT ( runReaderT ( runErrorT action ) w ) mos )
2010-06-21 15:06:20 +00:00
2010-11-01 00:38:38 +00:00
-- | A connection failure, or a read or write exception like cursor expired or inserting a duplicate key.
2010-07-27 21:18:53 +00:00
-- Note, unexpected data from the server is not a Failure, rather it is a programming error (you should call 'error' in this case) because the client and server are incompatible and requires a programming change.
2010-06-21 15:06:20 +00:00
data Failure =
2010-10-27 20:13:23 +00:00
ConnectionFailure IOError -- ^ TCP connection ('Pipe') failed. Make work if you try again on the same Mongo 'Connection' which will create a new Pipe.
| CursorNotFoundFailure CursorId -- ^ Cursor expired because it wasn't accessed for over 10 minutes, or this cursor came from a different server that the one you are currently connected to (perhaps a fail over happen between servers in a replica set)
2010-07-27 21:18:53 +00:00
| QueryFailure String -- ^ Query failed for some reason as described in the string
| WriteFailure ErrorCode String -- ^ Error observed by getLastError after a write, error description is in string
2010-06-21 15:06:20 +00:00
deriving ( Show , Eq )
2010-07-27 21:18:53 +00:00
instance Error Failure where strMsg = error
-- ^ 'fail' is treated the same as 'error'. In other words, don't use it.
2010-06-15 03:14:40 +00:00
-- * Database
2010-10-27 20:13:23 +00:00
newtype Database = Database { databaseName :: UString } deriving ( Eq , Ord )
2010-06-15 03:14:40 +00:00
-- ^ Database name
2010-10-27 20:13:23 +00:00
instance Show Database where show ( Database x ) = unpack x
2010-11-01 00:38:38 +00:00
-- | 'Access' monad with a particular 'Database' in context
2010-10-27 20:13:23 +00:00
class ( Context Database m , Access m ) => DbAccess m
instance ( Context Database m , Access m ) => DbAccess m
2010-06-21 15:06:20 +00:00
2010-10-27 20:13:23 +00:00
allDatabases :: ( Access m ) => m [ Database ]
2010-06-15 03:14:40 +00:00
-- ^ List all databases residing on server
2010-10-27 20:13:23 +00:00
allDatabases = map ( Database . at " name " ) . at " databases " <$> use ( Database " admin " ) ( runCommand1 " listDatabases " )
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
use :: Database -> ReaderT Database m a -> m a
2010-11-01 00:38:38 +00:00
-- ^ Run action against given database
2010-10-27 20:13:23 +00:00
use = flip runReaderT
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
thisDatabase :: ( DbAccess m ) => m Database
2010-06-15 03:14:40 +00:00
-- ^ Current database in use
2010-06-21 15:06:20 +00:00
thisDatabase = context
2010-06-15 03:14:40 +00:00
-- * Authentication
2010-10-27 20:13:23 +00:00
auth :: ( DbAccess m ) => Username -> Password -> m Bool
-- ^ Authenticate with the database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new pipe.
2010-12-27 05:23:02 +00:00
auth usr pss = do
2010-06-15 03:14:40 +00:00
n <- at " nonce " <$> runCommand [ " getnonce " =: ( 1 :: Int ) ]
2010-12-27 05:23:02 +00:00
true1 " ok " <$> runCommand [ " authenticate " =: ( 1 :: Int ) , " user " =: usr , " nonce " =: n , " key " =: pwKey n usr pss ]
2010-06-15 03:14:40 +00:00
-- * Collection
type Collection = UString
-- ^ Collection name (not prefixed with database)
2010-10-27 20:13:23 +00:00
allCollections :: ( DbAccess m ) => m [ Collection ]
2010-06-15 03:14:40 +00:00
-- ^ List all collections in this database
allCollections = do
db <- thisDatabase
docs <- rest =<< find ( query [] " system.namespaces " ) { sort = [ " name " =: ( 1 :: Int ) ] }
return . filter ( not . isSpecial db ) . map dropDbPrefix $ map ( at " name " ) docs
where
dropDbPrefix = U . tail . U . dropWhile ( /= '.' )
2010-10-27 20:13:23 +00:00
isSpecial ( Database db ) col = U . any ( == '$' ) col && db <.> col /= " local.oplog.$main "
2010-06-15 03:14:40 +00:00
-- * Selection
data Selection = Select { selector :: Selector , coll :: Collection } deriving ( Show , Eq )
-- ^ Selects documents in collection that match selector
type Selector = Document
-- ^ Filter for a query, analogous to the where clause in SQL. @[]@ matches all documents in collection. @[x =: a, y =: b]@ is analogous to @where x = a and y = b@ in SQL. See <http://www.mongodb.org/display/DOCS/Querying> for full selector syntax.
whereJS :: Selector -> Javascript -> Selector
-- ^ Add Javascript predicate to selector, in which case a document must match both selector and predicate
whereJS sel js = ( " $where " =: js ) : sel
2010-06-21 15:06:20 +00:00
class Select aQueryOrSelection where
select :: Selector -> Collection -> aQueryOrSelection
-- ^ 'Query' or 'Selection' that selects documents in collection that match selector. The choice of type depends on use, for example, in @find (select sel col)@ it is a Query, and in @delete (select sel col)@ it is a Selection.
instance Select Selection where
select = Select
instance Select Query where
select = query
2010-06-15 03:14:40 +00:00
-- * Write
2010-06-21 15:06:20 +00:00
-- ** WriteMode
-- | Default write-mode is 'Unsafe'
data WriteMode =
Unsafe -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.
2010-10-27 20:13:23 +00:00
| Safe GetLastError -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed. This is acomplished by sending the getLastError command, with given 'GetLastError' parameters, after every write.
2010-06-21 15:06:20 +00:00
deriving ( Show , Eq )
2010-10-27 20:13:23 +00:00
type GetLastError = Document
-- ^ Parameters for getLastError command. For example ["w" =: 2] tells the server to wait for the write to reach at least two servers in replica set before acknowledging. See "http://www.mongodb.org/display/DOCS/Last+Error+Commands" for more options.
safe :: WriteMode
-- ^ Safe []
safe = Safe []
writeMode :: ( Access m ) => WriteMode -> m a -> m a
2010-06-21 15:06:20 +00:00
-- ^ Run action with given 'WriteMode'
writeMode = push . const
2010-10-27 20:13:23 +00:00
write :: ( DbAccess m ) => Notice -> m ()
2010-07-27 21:18:53 +00:00
-- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'WriteFailure' if it reports an error.
2010-06-21 15:06:20 +00:00
write notice = do
mode <- context
case mode of
Unsafe -> send [ notice ]
2010-10-27 20:13:23 +00:00
Safe params -> do
me <- getLastError [ notice ] params
2010-07-27 21:18:53 +00:00
maybe ( return () ) ( throw . uncurry WriteFailure ) me
2010-06-21 15:06:20 +00:00
type ErrorCode = Int
-- ^ Error code from getLastError
2010-10-27 20:13:23 +00:00
getLastError :: ( DbAccess m ) => [ Notice ] -> GetLastError -> m ( Maybe ( ErrorCode , String ) )
2010-06-21 15:06:20 +00:00
-- ^ Send notices (writes) then fetch what the last error was, Nothing means no error
2010-10-27 20:13:23 +00:00
getLastError writes params = do
r <- runCommand' writes $ ( " getlasterror " =: ( 1 :: Int ) ) : params
2010-06-21 15:06:20 +00:00
return $ ( at " code " r , ) <$> lookup " err " r
{- resetLastError :: (DbConn m) => m ()
-- ^ Clear last error
resetLastError = runCommand1 " reseterror " >> return () - }
2010-06-15 03:14:40 +00:00
-- ** Insert
2010-10-27 20:13:23 +00:00
insert :: ( DbAccess m ) => Collection -> Document -> m Value
2010-06-15 03:14:40 +00:00
-- ^ Insert document into collection and return its \"_id\" value, which is created automatically if not supplied
insert col doc = head <$> insertMany col [ doc ]
2010-10-27 20:13:23 +00:00
insert_ :: ( DbAccess m ) => Collection -> Document -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Same as 'insert' except don't return _id
insert_ col doc = insert col doc >> return ()
2010-10-27 20:13:23 +00:00
insertMany :: ( DbAccess m ) => Collection -> [ Document ] -> m [ Value ]
2010-06-15 03:14:40 +00:00
-- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied
2010-06-21 15:06:20 +00:00
insertMany col docs = do
2010-10-27 20:13:23 +00:00
Database db <- thisDatabase
2010-06-15 03:14:40 +00:00
docs' <- liftIO $ mapM assignId docs
2010-06-21 15:06:20 +00:00
write ( Insert ( db <.> col ) docs' )
2010-06-15 03:14:40 +00:00
mapM ( look " _id " ) docs'
2010-10-27 20:13:23 +00:00
insertMany_ :: ( DbAccess m ) => Collection -> [ Document ] -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Same as 'insertMany' except don't return _ids
insertMany_ col docs = insertMany col docs >> return ()
assignId :: Document -> IO Document
-- ^ Assign a unique value to _id field if missing
assignId doc = if X . any ( ( " _id " == ) . label ) doc
then return doc
else ( \ oid -> ( " _id " =: oid ) : doc ) <$> genObjectId
-- ** Update
2010-10-27 20:13:23 +00:00
save :: ( DbAccess m ) => Collection -> Document -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Save document to collection, meaning insert it if its new (has no \"_id\" field) or update it if its not new (has \"_id\" field)
save col doc = case look " _id " doc of
Nothing -> insert_ col doc
Just i -> repsert ( Select [ " _id " := i ] col ) doc
2010-10-27 20:13:23 +00:00
replace :: ( DbAccess m ) => Selection -> Document -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Replace first document in selection with given document
replace = update []
2010-10-27 20:13:23 +00:00
repsert :: ( DbAccess m ) => Selection -> Document -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Replace first document in selection with given document, or insert document if selection is empty
repsert = update [ Upsert ]
type Modifier = Document
-- ^ Update operations on fields in a document. See <http://www.mongodb.org/display/DOCS/Updating#Updating-ModifierOperations>
2010-10-27 20:13:23 +00:00
modify :: ( DbAccess m ) => Selection -> Modifier -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Update all documents in selection using given modifier
modify = update [ MultiUpdate ]
2010-10-27 20:13:23 +00:00
update :: ( DbAccess m ) => [ UpdateOption ] -> Selection -> Document -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Update first document in selection using updater document, unless 'MultiUpdate' option is supplied then update all documents in selection. If 'Upsert' option is supplied then treat updater as document and insert it if selection is empty.
2010-06-21 15:06:20 +00:00
update opts ( Select sel col ) up = do
2010-10-27 20:13:23 +00:00
Database db <- thisDatabase
2010-06-21 15:06:20 +00:00
write ( Update ( db <.> col ) opts sel up )
2010-06-15 03:14:40 +00:00
-- ** Delete
2010-10-27 20:13:23 +00:00
delete :: ( DbAccess m ) => Selection -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Delete all documents in selection
2010-06-21 15:06:20 +00:00
delete = delete' []
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
deleteOne :: ( DbAccess m ) => Selection -> m ()
2010-06-15 03:14:40 +00:00
-- ^ Delete first document in selection
2010-06-21 15:06:20 +00:00
deleteOne = delete' [ SingleRemove ]
2010-10-27 20:13:23 +00:00
delete' :: ( DbAccess m ) => [ DeleteOption ] -> Selection -> m ()
2010-06-21 15:06:20 +00:00
-- ^ Delete all documents in selection unless 'SingleRemove' option is given then only delete first document in selection
delete' opts ( Select sel col ) = do
2010-10-27 20:13:23 +00:00
Database db <- thisDatabase
2010-06-21 15:06:20 +00:00
write ( Delete ( db <.> col ) opts sel )
2010-06-15 03:14:40 +00:00
-- * Read
2010-07-27 21:18:53 +00:00
-- ** MasterOrSlaveOk
2010-11-01 00:38:38 +00:00
readMode :: ( Access m ) => MasterOrSlaveOk -> m a -> m a
-- ^ Execute action using given read mode. Master = consistent reads, SlaveOk = eventually consistent reads.
readMode = push . const
2010-07-27 21:18:53 +00:00
msOption :: MasterOrSlaveOk -> [ P . QueryOption ]
msOption Master = []
msOption SlaveOk = [ P . SlaveOK ]
2010-06-15 03:14:40 +00:00
-- ** Query
2010-06-21 15:06:20 +00:00
-- | Use 'select' to create a basic query with defaults, then modify if desired. For example, @(select sel col) {limit = 10}@
2010-06-15 03:14:40 +00:00
data Query = Query {
2010-06-21 15:06:20 +00:00
options :: [ QueryOption ] , -- ^ Default = []
2010-06-15 03:14:40 +00:00
selection :: Selection ,
2010-06-21 15:06:20 +00:00
project :: Projector , -- ^ \[\] = all fields. Default = []
skip :: Word32 , -- ^ Number of initial matching documents to skip. Default = 0
limit :: Limit , -- ^ Maximum number of documents to return, 0 = no limit. Default = 0
sort :: Order , -- ^ Sort results by this order, [] = no sort. Default = []
snapshot :: Bool , -- ^ If true assures no duplicates are returned, or objects missed, which were present at both the start and end of the query's execution (even if the object were updated). If an object is new during the query, or deleted during the query, it may or may not be returned, even with snapshot mode. Note that short query responses (less than 1MB) are always effectively snapshotted. Default = False
batchSize :: BatchSize , -- ^ The number of document to return in each batch response from the server. 0 means use Mongo default. Default = 0
hint :: Order -- ^ Force MongoDB to use this index, [] = no hint. Default = []
2010-06-15 03:14:40 +00:00
} deriving ( Show , Eq )
2010-07-27 21:18:53 +00:00
data QueryOption =
TailableCursor -- ^ Tailable means cursor is not closed when the last data is retrieved. Rather, the cursor marks the final object's position. You can resume using the cursor later, from where it was located, if more data were received. Like any "latent cursor", the cursor may become invalid at some point – for example if the final object it references were deleted. Thus, you should be prepared to requery on CursorNotFound exception.
| NoCursorTimeout -- The server normally times out idle cursors after an inactivity period (10 minutes) to prevent excess memory use. Set this option to prevent that.
| AwaitData -- ^ Use with TailableCursor. If we are at the end of the data, block for a while rather than returning no data. After a timeout period, we do return as normal.
deriving ( Show , Eq )
pOption :: QueryOption -> P . QueryOption
-- ^ Convert to protocol query option
pOption TailableCursor = P . TailableCursor
pOption NoCursorTimeout = P . NoCursorTimeout
pOption AwaitData = P . AwaitData
2010-06-15 03:14:40 +00:00
type Projector = Document
-- ^ Fields to return, analogous to the select clause in SQL. @[]@ means return whole document (analogous to * in SQL). @[x =: 1, y =: 1]@ means return only @x@ and @y@ fields of each document. @[x =: 0]@ means return all fields except @x@.
type Limit = Word32
-- ^ Maximum number of documents to return, i.e. cursor will close after iterating over this number of documents. 0 means no limit.
type Order = Document
2010-07-03 17:15:30 +00:00
-- ^ Fields to sort by. Each one is associated with 1 or -1. Eg. @[x =: 1, y =: -1]@ means sort by @x@ ascending then @y@ descending
2010-06-15 03:14:40 +00:00
type BatchSize = Word32
-- ^ The number of document to return in each batch response from the server. 0 means use Mongo default.
query :: Selector -> Collection -> Query
-- ^ Selects documents in collection that match selector. It uses no query options, projects all fields, does not skip any documents, does not limit result size, uses default batch size, does not sort, does not hint, and does not snapshot.
query sel col = Query [] ( Select sel col ) [] 0 0 [] False 0 []
batchSizeRemainingLimit :: BatchSize -> Limit -> ( Int32 , Limit )
-- ^ Given batchSize and limit return P.qBatchSize and remaining limit
batchSizeRemainingLimit batchSize limit = if limit == 0
2010-07-03 17:15:30 +00:00
then ( fromIntegral batchSize' , 0 ) -- no limit
else if 0 < batchSize' && batchSize' < limit
then ( fromIntegral batchSize' , limit - batchSize' )
2010-06-15 03:14:40 +00:00
else ( - fromIntegral limit , 1 )
2010-07-03 17:15:30 +00:00
where batchSize' = if batchSize == 1 then 2 else batchSize
-- batchSize 1 is broken because server converts 1 to -1 meaning limit 1
2010-06-15 03:14:40 +00:00
2010-07-27 21:18:53 +00:00
queryRequest :: Bool -> MasterOrSlaveOk -> Query -> Database -> ( Request , Limit )
2010-06-15 03:14:40 +00:00
-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.
2010-10-27 20:13:23 +00:00
queryRequest isExplain mos Query { .. } ( Database db ) = ( P . Query { .. } , remainingLimit ) where
2010-07-27 21:18:53 +00:00
qOptions = msOption mos ++ map pOption options
2010-06-15 03:14:40 +00:00
qFullCollection = db <.> coll selection
qSkip = fromIntegral skip
( qBatchSize , remainingLimit ) = batchSizeRemainingLimit batchSize limit
qProjector = project
mOrder = if null sort then Nothing else Just ( " $orderby " =: sort )
mSnapshot = if snapshot then Just ( " $snapshot " =: True ) else Nothing
mHint = if null hint then Nothing else Just ( " $hint " =: hint )
mExplain = if isExplain then Just ( " $explain " =: True ) else Nothing
special = catMaybes [ mOrder , mSnapshot , mHint , mExplain ]
qSelector = if null special then s else ( " $query " =: s ) : special where s = selector selection
2010-11-01 19:35:13 +00:00
runQuery :: ( DbAccess m ) => Bool -> [ Notice ] -> Query -> m DelayedCursorState
2010-06-21 15:06:20 +00:00
-- ^ Send query request and return cursor state
2010-07-27 21:18:53 +00:00
runQuery isExplain ns q = do
db <- thisDatabase
2010-07-27 21:30:52 +00:00
slaveOK <- context
2010-11-01 19:35:13 +00:00
request ns ( queryRequest isExplain slaveOK q db )
2010-07-03 17:15:30 +00:00
2010-10-27 20:13:23 +00:00
find :: ( DbAccess m ) => Query -> m Cursor
2010-06-15 03:14:40 +00:00
-- ^ Fetch documents satisfying query
2010-06-21 15:06:20 +00:00
find q @ Query { selection , batchSize } = do
db <- thisDatabase
2010-11-01 19:35:13 +00:00
dcs <- runQuery False [] q
newCursor db ( coll selection ) batchSize dcs
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
findOne' :: ( DbAccess m ) => [ Notice ] -> Query -> m ( Maybe Document )
2010-06-21 15:06:20 +00:00
-- ^ Send notices and fetch first document satisfying query or Nothing if none satisfy it
findOne' ns q = do
2010-11-01 19:35:13 +00:00
CS _ _ docs <- mapErrorIO id =<< runQuery False ns q { limit = 1 }
2010-06-15 03:14:40 +00:00
return ( listToMaybe docs )
2010-10-27 20:13:23 +00:00
findOne :: ( DbAccess m ) => Query -> m ( Maybe Document )
2010-06-21 15:06:20 +00:00
-- ^ Fetch first document satisfying query or Nothing if none satisfy it
findOne = findOne' []
2010-10-27 20:13:23 +00:00
explain :: ( DbAccess m ) => Query -> m Document
2010-06-15 03:14:40 +00:00
-- ^ Return performance stats of query execution
2010-06-21 15:06:20 +00:00
explain q = do -- same as findOne but with explain set to true
2010-11-01 19:35:13 +00:00
CS _ _ docs <- mapErrorIO id =<< runQuery True [] q { limit = 1 }
2010-06-21 15:06:20 +00:00
return $ if null docs then error ( " no explain: " ++ show q ) else head docs
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
count :: ( DbAccess m ) => Query -> m Int
2010-06-15 03:14:40 +00:00
-- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present)
count Query { selection = Select sel col , skip , limit } = at " n " <$> runCommand
( [ " count " =: col , " query " =: sel , " skip " =: ( fromIntegral skip :: Int32 ) ]
++ ( " limit " =? if limit == 0 then Nothing else Just ( fromIntegral limit :: Int32 ) ) )
2010-10-27 20:13:23 +00:00
distinct :: ( DbAccess m ) => Label -> Selection -> m [ Value ]
2010-06-15 03:14:40 +00:00
-- ^ Fetch distinct values of field in selected documents
distinct k ( Select sel col ) = at " values " <$> runCommand [ " distinct " =: col , " key " =: k , " query " =: sel ]
-- *** Cursor
2010-11-01 19:35:13 +00:00
data Cursor = Cursor FullCollection BatchSize ( MVar DelayedCursorState )
2010-10-27 20:13:23 +00:00
-- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor.
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
getCursorState :: ( Access m ) => Cursor -> m CursorState
2010-06-21 15:06:20 +00:00
-- ^ Extract current cursor status
2010-11-01 19:35:13 +00:00
getCursorState ( Cursor _ _ var ) = mapErrorIO id =<< readMVar var
2010-06-21 15:06:20 +00:00
2010-11-01 19:35:13 +00:00
type DelayedCursorState = ErrorT Failure IO CursorState
-- ^ A promised cursor state which may fail
2010-06-21 15:06:20 +00:00
2010-11-01 19:35:13 +00:00
request :: ( Access m ) => [ Notice ] -> ( Request , Limit ) -> m DelayedCursorState
2010-07-27 21:18:53 +00:00
-- ^ Send notices and request and return promised cursor state
2010-11-01 19:35:13 +00:00
request ns ( req , remainingLimit ) = do
2010-07-27 21:18:53 +00:00
promise <- call ns req
2010-11-01 19:35:13 +00:00
return $ fromReply remainingLimit =<< promise
2010-06-21 15:06:20 +00:00
2010-06-15 03:14:40 +00:00
data CursorState = CS Limit CursorId [ Document ]
-- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch.
2010-07-27 21:18:53 +00:00
fromReply :: ( Throw Failure m ) => Limit -> Reply -> m CursorState
2010-06-21 15:06:20 +00:00
-- ^ Convert Reply to CursorState or Failure
2010-07-27 21:18:53 +00:00
fromReply limit Reply { .. } = do
mapM_ checkResponseFlag rResponseFlags
return ( CS limit rCursorId rDocuments )
2010-07-03 17:15:30 +00:00
where
2010-07-27 21:18:53 +00:00
-- If response flag indicates failure then throw it, otherwise do nothing
checkResponseFlag flag = case flag of
AwaitCapable -> return ()
CursorNotFound -> throw ( CursorNotFoundFailure rCursorId )
QueryError -> throw ( QueryFailure $ at " $err " $ head rDocuments )
2010-06-15 03:14:40 +00:00
2010-11-01 19:35:13 +00:00
newCursor :: ( Access m ) => Database -> Collection -> BatchSize -> DelayedCursorState -> m Cursor
2010-06-21 15:06:20 +00:00
-- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected.
2010-10-27 20:13:23 +00:00
newCursor ( Database db ) col batch cs = do
2010-11-01 19:35:13 +00:00
var <- newMVar cs
2010-06-21 15:06:20 +00:00
let cursor = Cursor ( db <.> col ) batch var
2010-12-20 02:08:53 +00:00
addMVarFinalizer var ( closeCursor cursor )
2010-06-21 15:06:20 +00:00
return cursor
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
next :: ( Access m ) => Cursor -> m ( Maybe Document )
2010-06-15 20:15:37 +00:00
-- ^ Return next document in query result, or Nothing if finished.
2010-11-01 19:35:13 +00:00
next ( Cursor fcol batch var ) = modifyMVar var nextState where
2010-06-21 15:06:20 +00:00
-- Pre-fetch next batch promise from server when last one in current batch is returned.
2010-11-01 19:35:13 +00:00
nextState dcs = do
CS limit cid docs <- mapErrorIO id dcs
2010-06-21 15:06:20 +00:00
case docs of
doc : docs' -> do
2010-11-01 19:35:13 +00:00
dcs' <- if null docs' && cid /= 0
then nextBatch limit cid
else return $ return ( CS limit cid docs' )
return ( dcs' , Just doc )
2010-06-21 15:06:20 +00:00
[] -> if cid == 0
2010-11-01 19:35:13 +00:00
then return ( return $ CS 0 0 [] , Nothing ) -- finished
2010-06-21 15:06:20 +00:00
else error $ " server returned empty batch but says more results on server "
2010-11-01 19:35:13 +00:00
nextBatch limit cid = request [] ( GetMore fcol batchSize cid , remLimit )
2010-07-03 17:15:30 +00:00
where ( batchSize , remLimit ) = batchSizeRemainingLimit batch limit
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
nextN :: ( Access m ) => Int -> Cursor -> m [ Document ]
2010-06-15 03:14:40 +00:00
-- ^ Return next N documents or less if end is reached
nextN n c = catMaybes <$> replicateM n ( next c )
2010-10-27 20:13:23 +00:00
rest :: ( Access m ) => Cursor -> m [ Document ]
2010-06-15 03:14:40 +00:00
-- ^ Return remaining documents in query result
rest c = loop ( next c )
2010-12-20 02:08:53 +00:00
closeCursor :: ( Access m ) => Cursor -> m ()
closeCursor ( Cursor _ _ var ) = modifyMVar var kill' where
kill' dcs = first return <$> ( kill =<< mapErrorIO id dcs )
kill ( CS _ cid _ ) = ( CS 0 0 [] , ) <$> if cid == 0 then return () else send [ KillCursors [ cid ] ]
isCursorClosed :: ( Access m ) => Cursor -> m Bool
isCursorClosed cursor = do
2010-06-21 15:06:20 +00:00
CS _ cid docs <- getCursorState cursor
return ( cid == 0 && null docs )
2010-06-15 03:14:40 +00:00
-- ** Group
2010-07-03 17:15:30 +00:00
-- | Groups documents in collection by key then reduces (aggregates) each group
2010-06-15 03:14:40 +00:00
data Group = Group {
gColl :: Collection ,
gKey :: GroupKey , -- ^ Fields to group by
2010-07-03 17:15:30 +00:00
gReduce :: Javascript , -- ^ @(doc, agg) -> ()@. The reduce function reduces (aggregates) the objects iterated. Typical operations of a reduce function include summing and counting. It takes two arguments, the current document being iterated over and the aggregation value, and updates the aggregate value.
gInitial :: Document , -- ^ @agg@. Initial aggregation value supplied to reduce
2010-06-15 03:14:40 +00:00
gCond :: Selector , -- ^ Condition that must be true for a row to be considered. [] means always true.
2010-07-03 17:15:30 +00:00
gFinalize :: Maybe Javascript -- ^ @agg -> () | result@. An optional function to be run on each item in the result set just before the item is returned. Can either modify the item (e.g., add an average field given a count and a total) or return a replacement object (returning a new object with just _id and average fields).
2010-06-15 03:14:40 +00:00
} deriving ( Show , Eq )
data GroupKey = Key [ Label ] | KeyF Javascript deriving ( Show , Eq )
2010-07-03 17:15:30 +00:00
-- ^ Fields to group by, or function (@doc -> key@) returning a "key object" to be used as the grouping key. Use KeyF instead of Key to specify a key that is not an existing member of the object (or, to access embedded members).
2010-06-15 03:14:40 +00:00
groupDocument :: Group -> Document
-- ^ Translate Group data into expected document form
groupDocument Group { .. } =
( " finalize " =? gFinalize ) ++ [
" ns " =: gColl ,
case gKey of Key k -> " key " =: map ( =: True ) k ; KeyF f -> " $keyf " =: f ,
" $reduce " =: gReduce ,
" initial " =: gInitial ,
" cond " =: gCond ]
2010-10-27 20:13:23 +00:00
group :: ( DbAccess m ) => Group -> m [ Document ]
2010-06-15 03:14:40 +00:00
-- ^ Execute group query and return resulting aggregate value for each distinct key
group g = at " retval " <$> runCommand [ " group " =: groupDocument g ]
-- ** MapReduce
2010-10-27 20:13:23 +00:00
-- | Maps every document in collection to a list of (key, value) pairs, then for each unique key reduces all its associated values to a single result. There are additional parameters that may be set to tweak this basic operation.
2010-06-15 03:14:40 +00:00
data MapReduce = MapReduce {
rColl :: Collection ,
rMap :: MapFun ,
rReduce :: ReduceFun ,
2010-07-03 17:15:30 +00:00
rSelect :: Selector , -- ^ Operate on only those documents selected. Default is [] meaning all documents.
2010-06-15 03:14:40 +00:00
rSort :: Order , -- ^ Default is [] meaning no sort
rLimit :: Limit , -- ^ Default is 0 meaning no limit
2010-06-15 20:15:37 +00:00
rOut :: Maybe Collection , -- ^ Output to given permanent collection, otherwise output to a new temporary collection whose name is returned.
2010-10-27 20:13:23 +00:00
rKeepTemp :: Bool , -- ^ If True, the temporary output collection is made permanent. If False, the temporary output collection persists for the life of the current pipe only, however, other pipes may read from it while the original one is still alive. Note, reading from a temporary collection after its original pipe dies returns an empty result (not an error). The default for this attribute is False, unless 'rOut' is specified, then the collection permanent.
2010-06-15 03:14:40 +00:00
rFinalize :: Maybe FinalizeFun , -- ^ Function to apply to all the results when finished. Default is Nothing.
rScope :: Document , -- ^ Variables (environment) that can be accessed from map/reduce/finalize. Default is [].
rVerbose :: Bool -- ^ Provide statistics on job execution time. Default is False.
} deriving ( Show , Eq )
type MapFun = Javascript
2010-07-03 17:15:30 +00:00
-- ^ @() -> void@. The map function references the variable @this@ to inspect the current object under consideration. The function must call @emit(key,value)@ at least once, but may be invoked any number of times, as may be appropriate.
2010-06-15 03:14:40 +00:00
type ReduceFun = Javascript
2010-10-27 20:13:23 +00:00
-- ^ @(key, [value]) -> value@. The reduce function receives a key and an array of values and returns an aggregate result value. The MapReduce engine may invoke reduce functions iteratively; thus, these functions must be idempotent. That is, the following must hold for your reduce function: @reduce(k, [reduce(k,vs)]) == reduce(k,vs)@. If you need to perform an operation only once, use a finalize function. The output of emit (the 2nd param) and reduce should be the same format to make iterative reduce possible.
2010-06-15 03:14:40 +00:00
type FinalizeFun = Javascript
-- ^ @(key, value) -> final_value@. A finalize function may be run after reduction. Such a function is optional and is not necessary for many map/reduce cases. The finalize function takes a key and a value, and returns a finalized value.
mrDocument :: MapReduce -> Document
-- ^ Translate MapReduce data into expected document form
mrDocument MapReduce { .. } =
( " mapreduce " =: rColl ) :
( " out " =? rOut ) ++
( " finalize " =? rFinalize ) ++ [
" map " =: rMap ,
" reduce " =: rReduce ,
" query " =: rSelect ,
" sort " =: rSort ,
" limit " =: ( fromIntegral rLimit :: Int ) ,
" keeptemp " =: rKeepTemp ,
" scope " =: rScope ,
" verbose " =: rVerbose ]
mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce
-- ^ MapReduce on collection with given map and reduce functions. Remaining attributes are set to their defaults, which are stated in their comments.
mapReduce col map' red = MapReduce col map' red [] [] 0 Nothing False Nothing [] False
2010-10-27 20:13:23 +00:00
runMR :: ( DbAccess m ) => MapReduce -> m Cursor
2010-06-15 03:14:40 +00:00
-- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript)
2010-10-27 20:13:23 +00:00
-- TODO: Delete temp result collection when cursor closes. Until then, it will be deleted by the server when pipe closes.
2010-06-15 03:14:40 +00:00
runMR mr = find . query [] =<< ( at " result " <$> runMR' mr )
2010-10-27 20:13:23 +00:00
runMR' :: ( DbAccess m ) => MapReduce -> m Document
2010-06-15 03:14:40 +00:00
-- ^ Run MapReduce and return a result document containing a "result" field holding the output Collection and additional statistic fields. Error if the map/reduce failed (because of bad Javascript).
runMR' mr = do
doc <- runCommand ( mrDocument mr )
2010-10-27 20:13:23 +00:00
return $ if true1 " ok " doc then doc else error $ " mapReduce error: \ n " ++ show doc ++ " \ n in: \ n " ++ show mr
2010-06-15 03:14:40 +00:00
-- * Command
type Command = Document
-- ^ A command is a special query or action against the database. See <http://www.mongodb.org/display/DOCS/Commands> for details.
2010-10-27 20:13:23 +00:00
runCommand' :: ( DbAccess m ) => [ Notice ] -> Command -> m Document
2010-06-21 15:06:20 +00:00
-- ^ Send notices then run command and return its result
runCommand' ns c = maybe err id <$> findOne' ns ( query c " $cmd " ) where
err = error $ " Nothing returned for command: " ++ show c
2010-10-27 20:13:23 +00:00
runCommand :: ( DbAccess m ) => Command -> m Document
2010-06-15 03:14:40 +00:00
-- ^ Run command against the database and return its result
2010-06-21 15:06:20 +00:00
runCommand = runCommand' []
2010-06-15 03:14:40 +00:00
2010-10-27 20:13:23 +00:00
runCommand1 :: ( DbAccess m ) => UString -> m Document
2010-06-21 15:06:20 +00:00
-- ^ @runCommand1 foo = runCommand [foo =: 1]@
2010-06-15 03:14:40 +00:00
runCommand1 c = runCommand [ c =: ( 1 :: Int ) ]
2010-10-27 20:13:23 +00:00
eval :: ( DbAccess m ) => Javascript -> m Document
2010-06-15 03:14:40 +00:00
-- ^ Run code on server
eval code = at " retval " <$> runCommand [ " $eval " =: code ]
2010-11-01 00:38:38 +00:00
-- * Primitives
send :: ( Context Pipe m , Throw Failure m , MonadIO m ) => [ Notice ] -> m ()
-- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.
2010-11-01 19:35:13 +00:00
send ns = do
pipe <- context
mapErrorIO ConnectionFailure ( P . send pipe ns )
2010-11-01 00:38:38 +00:00
2010-12-20 02:08:53 +00:00
call :: ( Context Pipe m , Throw Failure m , MonadIO m , Throw Failure n , MonadIO n ) =>
[ Notice ] -> Request -> m ( n Reply )
2010-11-01 00:38:38 +00:00
-- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call will throw 'ConnectionFailure' if pipe fails on send, and promise will throw 'ConnectionFailure' if pipe fails on receive.
call ns r = do
pipe <- context
promise <- mapErrorIO ConnectionFailure ( P . call pipe ns r )
return ( mapErrorIO ConnectionFailure promise )
2010-06-15 03:14:40 +00:00
{- Authors: Tony Hannan <tony@10gen.com>
Copyright 2010 10 gen Inc .
Licensed under the Apache License , Version 2.0 ( the " License " ) ; you may not use this file except in compliance with the License . You may obtain a copy of the License at : http :// www . apache . org / licenses / LICENSE - 2.0 . Unless required by applicable law or agreed to in writing , software distributed under the License is distributed on an " AS IS " BASIS , WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND , either express or implied . See the License for the specific language governing permissions and limitations under the License . - }