Merge pull request #60 from VictorDenisov/delete_command

Delete command

CHANGELOG.md

@@ -9,11 +9,16 @@ This project adheres to [Package Versioning Policy](https://wiki.haskell.org/Pac
 - Insert using command syntax with mongo server >= 2.6
 - UpdateMany and UpdateAll commands. They use bulk operations from mongo
   version 2.6 and above. With versions below 2.6 it sends many updates.
+- DeleteAll and DeleteMany functions use bulk operations with mongo server
+  >= 2.6. If mongo server version is below 2.6 then it sends many individual
+  deletes.
 
 ### Changed
 - All messages will be strictly evaluated before sending them to mongodb server.
 No more closed handles because of bad arguments.
 - Update command is reimplemented in terms of UpdateMany.
+- delete and deleteOne functions are now implemented using bulk delete
+  functions.
 
 ### Removed
 - System.IO.Pipeline module

Database/MongoDB/Admin.hs

@@ -204,7 +204,8 @@ addUser readOnly user pass = do
     let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu)
     save "system.users" usr
 
-removeUser :: (MonadIO m) => Username -> Action m ()
+removeUser :: (MonadIO m, MonadBaseControl IO m)
+           => Username -> Action m ()
 removeUser user = delete (select ["user" =: user] "system.users")
 
 -- ** Database

Database/MongoDB/Query.hs

@@ -43,7 +43,8 @@ module Database.MongoDB.Query (
     -- * Command
     Command, runCommand, runCommand1,
     eval, retrieveServerData, updateMany, updateAll, UpdateResult,
-    UpdateOption(..)
+    UpdateOption(..),
+    deleteMany, deleteAll, DeleteResult, DeleteOption(..)
 ) where
 
 import Prelude hiding (lookup)
@@ -150,6 +151,8 @@ type GetLastError = Document
 
 data UpdateResult = UpdateResult
 
+data DeleteResult = DeleteResult
+
 master :: AccessMode
 -- ^ Same as 'ConfirmWrites' []
 master = ConfirmWrites []
@@ -440,7 +443,7 @@ insert' opts col docs = do
   chunks <- forM (splitAtLimit
                       (not (KeepGoing `elem` opts))
                       (maxBsonObjectSize sd - docSize)
-                                           -- ^ size of auxiliary part of insert
+                                           -- size of auxiliary part of insert
                                            -- document should be subtracted from
                                            -- the overall size
                       (maxWriteBatchSize sd)
@@ -609,7 +612,7 @@ update' ordered col updateDocs = do
   let chunks = splitAtLimit
                       ordered
                       (maxBsonObjectSize sd - docSize)
-                                           -- ^ size of auxiliary part of insert
+                                           -- size of auxiliary part of update
                                            -- document should be subtracted from
                                            -- the overall size
                       (maxWriteBatchSize sd)
@@ -663,19 +666,122 @@ updateBlock ordered col docs = do
 
 -- ** Delete
 
-delete :: (MonadIO m) => Selection -> Action m ()
+delete :: (MonadIO m, MonadBaseControl IO m)
+       => Selection -> Action m ()
 -- ^ Delete all documents in selection
-delete = delete' []
+delete = deleteHelper []
 
-deleteOne :: (MonadIO m) => Selection -> Action m ()
+deleteOne :: (MonadIO m, MonadBaseControl IO m)
+          => Selection -> Action m ()
 -- ^ Delete first document in selection
-deleteOne = delete' [SingleRemove]
+deleteOne = deleteHelper [SingleRemove]
 
-delete' :: (MonadIO m) => [DeleteOption] -> Selection -> Action m ()
+deleteHelper :: (MonadBaseControl IO m, MonadIO m)
--- ^ Delete all documents in selection unless 'SingleRemove' option is given then only delete first document in selection
+             => [DeleteOption] -> Selection -> Action m ()
-delete' opts (Select sel col) = do
+deleteHelper opts (Select sel col) = void $ delete' True col [(sel, opts)]
-    db <- thisDatabase
+
-    write (Delete (db <.> col) opts sel)
+{-| Bulk delete operation. If one delete fails it will not delete the remaining
+ - documents. Current returned value is only a place holder. With mongodb server
+ - before 2.6 it will send delete requests one by one. After 2.6 it will use
+ - bulk delete feature in mongodb.
+ -}
+deleteMany :: (MonadIO m, MonadBaseControl IO m)
+           => Collection
+           -> [(Selector, [DeleteOption])]
+           -> Action m DeleteResult
+deleteMany = delete' True
+
+{-| Bulk delete operation. If one delete fails it will proceed with the
+ - remaining documents. Current returned value is only a place holder. With
+ - mongodb server before 2.6 it will send delete requests one by one. After 2.6
+ - it will use bulk delete feature in mongodb.
+ -}
+deleteAll :: (MonadIO m, MonadBaseControl IO m)
+          => Collection
+          -> [(Selector, [DeleteOption])]
+          -> Action m DeleteResult
+deleteAll = delete' False
+
+deleteCommandDocument :: Collection -> Bool -> [Document] -> Document -> Document
+deleteCommandDocument col ordered deletes writeConcern =
+  [ "delete"       =: col
+  , "ordered"      =: ordered
+  , "deletes"      =: deletes
+  , "writeConcern" =: writeConcern
+  ]
+
+delete' :: (MonadIO m, MonadBaseControl IO m)
+        => Bool
+        -> Collection
+        -> [(Selector, [DeleteOption])]
+        -> Action m DeleteResult
+delete' ordered col deleteDocs = do
+  p <- asks mongoPipe
+  let sd = P.serverData p
+  let deletes = map (\(s, os) -> [ "q"     =: s
+                                 , "limit" =: if SingleRemove `elem` os
+                                               then (1 :: Int) -- Remove only one matching
+                                               else (0 :: Int) -- Remove all matching
+                                 ])
+                    deleteDocs
+
+  mode <- asks mongoWriteMode
+  let writeConcern = case mode of
+                        NoConfirm -> ["w" =: (0 :: Int)]
+                        Confirm params -> params
+  let docSize = sizeOfDocument $ deleteCommandDocument col ordered [] writeConcern
+  let chunks = splitAtLimit
+                      ordered
+                      (maxBsonObjectSize sd - docSize)
+                                           -- size of auxiliary part of delete
+                                           -- document should be subtracted from
+                                           -- the overall size
+                      (maxWriteBatchSize sd)
+                      deletes
+  forM_ chunks (deleteBlock ordered col)
+  return DeleteResult
+
+deleteBlock :: (MonadIO m, MonadBaseControl IO m)
+            => Bool -> Collection -> [Document] -> Action m ()
+deleteBlock ordered col docs = do
+  p <- asks mongoPipe
+  let sd = P.serverData p
+  if (maxWireVersion sd < 2)
+    then do
+      db <- thisDatabase
+      errors <-
+        forM docs $ \deleteDoc -> do
+          let sel = (at "q" deleteDoc) :: Document
+          let opts = if at "limit" deleteDoc == (1 :: Int) then [SingleRemove] else []
+          write (Delete (db <.> col) opts sel)
+          return Nothing
+          `catch` \(e :: SomeException) -> do
+                     when ordered $ liftIO $ throwIO e
+                     return $ Just e
+      let onlyErrors = catMaybes errors
+      if not $ null onlyErrors
+        then liftIO $ throwIO $ WriteFailure 0 (show onlyErrors)
+        else return ()
+    else do
+      mode <- asks mongoWriteMode
+      let writeConcern = case mode of
+                          NoConfirm -> ["w" =: (0 :: Int)]
+                          Confirm params -> params
+      doc <- runCommand $ deleteCommandDocument col ordered docs writeConcern
+      case (look "writeErrors" doc, look "writeConcernError" doc) of
+        (Nothing, Nothing) -> return ()
+        (Just err, Nothing) -> do
+          liftIO $ throwIO $ WriteFailure
+                                  (maybe 0 id $ lookup "ok" doc)
+                                  (show err)
+        (Nothing, Just err) -> do
+          liftIO $ throwIO $ WriteFailure
+                                  (maybe 0 id $ lookup "ok" doc)
+                                  (show err)
+        (Just err, Just writeConcernErr) -> do
+          liftIO $ throwIO $ WriteFailure
+                                  (maybe 0 id $ lookup "ok" doc)
+                                  (show err ++ show writeConcernErr)
 
 -- * Read
 

test/QuerySpec.hs

@@ -261,7 +261,7 @@ spec = around withCleanDatabase $ do
                                                      , ["league" =: "MiLB"    , "name" =: "Giants" , "score" =: (1 :: Int)]
                                                      ]
     it "can handle big updates" $ do
-      let docs = (flip map) [0..200000] $ \i ->
+      let docs = (flip map) [0..20000] $ \i ->
               ["name" =: (T.pack $ "name " ++ (show i))]
       ids <- db $ insertAll "bigCollection" docs
       let updateDocs = (flip map) ids (\i -> ( [ "_id" =: i]
@@ -291,6 +291,75 @@ spec = around withCleanDatabase $ do
                                                      , ["league" =: "MiLB"    , "name" =: "Giants" , "score" =: (3 :: Int)]
                                                      ]
 
+  describe "delete" $ do
+    it "actually deletes something" $ do
+      _ <- db $ insert "team" ["name" =: ("Giants" :: String)]
+      db $ delete $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+    it "deletes all matching entries" $ do
+      _ <- db $ insert "team" ["name" =: ("Giants" :: String)]
+      _ <- db $ insert "team" [ "name"  =: ("Giants" :: String)
+                              , "score" =: (10 :: Int)
+                              ]
+      db $ delete $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+    it "works if there is no matching document" $ do
+      db $ delete $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+
+  describe "deleteOne" $ do
+    it "actually deletes something" $ do
+      _ <- db $ insert "team" ["name" =: ("Giants" :: String)]
+      db $ deleteOne $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+    it "deletes only one matching entry" $ do
+      _ <- db $ insert "team" ["name" =: ("Giants" :: String)]
+      _ <- db $ insert "team" [ "name"  =: ("Giants" :: String)
+                              , "score" =: (10 :: Int)
+                              ]
+      db $ deleteOne $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 1
+    it "works if there is no matching document" $ do
+      db $ deleteOne $ select ["name" =: "Giants"] "team"
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+
+  describe "deleteMany" $ do
+    it "actually deletes something" $ do
+      _ <- db $ insert "team" ["name" =: ("Giants" :: String)]
+      _ <- db $ insert "team" ["name" =: ("Yankees" :: String)]
+      _ <- db $ deleteMany "team" [ (["name" =: ("Giants" :: String)], [])
+                                  , (["name" =: ("Yankees" :: String)], [])
+                                  ]
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+
+  describe "deleteAll" $ do
+    it "actually deletes something" $ do
+      _ <- db $ insert "team" [ "name"  =: ("Giants" :: String)
+                              , "score" =: (Nothing :: Maybe Int)
+                              ]
+      _ <- db $ insert "team" [ "name" =: ("Yankees" :: String)
+                              , "score" =: (1 :: Int)
+                              ]
+      _ <- db $ deleteAll "team" [ (["name" =: ("Giants" :: String)], [])
+                                 , (["name" =: ("Yankees" :: String)], [])
+                                 ]
+      updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+    it "can handle big deletes" $ do
+      let docs = (flip map) [0..20000] $ \i ->
+              ["name" =: (T.pack $ "name " ++ (show i))]
+      _ <- db $ insertAll "bigCollection" docs
+      _ <- db $ deleteAll "bigCollection" $ map (\d -> (d, [])) docs
+      updatedResult <- db $ rest =<< find ((select [] "bigCollection") {project = ["_id" =: (0 :: Int)]})
+      length updatedResult `shouldBe` 0
+
   describe "allCollections" $ do
     it "returns all collections in a database" $ do
       _ <- db $ insert "team1" ["name" =: "Yankees", "league" =: "American"]