2d8be0d · typeable/bloodhound

git.iko.soy/typeable/bloodhound

on master help sign in

Add optimistic concurrency control version feature

2d8be0dMichael Xavier11 years ago

src/Database/Bloodhound/Client.hs+41−4

compare with master

unifiedsplit

@@ -47,6 +47,10 @@ module Database.Bloodhound.Client

47 , getStatus

48 , encodeBulkOperations

49 , encodeBulkOperation

50+ -- * Reply-handling tools

51+ , isVersionConflict

52+ , isSuccess

53+ , isCreated

50 )

54 )

51 where

55 where

@@ -58,7 +62,9 @@ import Data.Aeson

58import Data.ByteString.Lazy.Builder

62import Data.ByteString.Lazy.Builder

59import qualified Data.ByteString.Lazy.Char8 as L

63import qualified Data.ByteString.Lazy.Char8 as L

60import Data.Default.Class

64import Data.Default.Class

65+import Data.Ix

61import Data.Maybe (fromMaybe)

66import Data.Maybe (fromMaybe)

67+import Data.Monoid

62import Data.Text (Text)

68import Data.Text (Text)

63import qualified Data.Text as T

69import qualified Data.Text as T

64import qualified Data.Text.Encoding as T

70import qualified Data.Text.Encoding as T

@@ -160,6 +166,15 @@ joinPath ps = do

160 Server s <- bhServer <$> getBHEnv

166 Server s <- bhServer <$> getBHEnv

161 return $ joinPath' (s:ps)

167 return $ joinPath' (s:ps)

162

168

169+-- | Severely dumbed down query renderer. Assumes your data doesn't

170+-- need any encoding

171+addQuery :: [(Text, Text)] -> Text -> Text

172+addQuery ps u = u <> "?" <> params

173+ where

174+ params = T.intercalate "&" (uncurry mkParam <$> ps)

175+ mkParam k v = k <> "=" <> v

176+

177+

163bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c

178bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c

164bindM2 f ma mb = join (f <$> ma <*> mb)

179bindM2 f ma mb = join (f <$> ma <*> mb)

165

180

@@ -315,15 +330,23 @@ deleteMapping (IndexName indexName)

315-- convert into a JSON 'Value'. The 'DocId' will function as the

330-- convert into a JSON 'Value'. The 'DocId' will function as the

316-- primary key for the document.

331-- primary key for the document.

317--

332--

318−-- >>> resp <- runBH' $ indexDocument testIndex testMapping exampleTweet (DocId "1")

333+-- >>> resp <- runBH' $ indexDocument defaultIndexDocumentSettings testIndex testMapping exampleTweet (DocId "1")

319-- >>> print resp

334-- >>> print resp

320-- Response {responseStatus = Status {statusCode = 201, statusMessage = "Created"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","74")], responseBody = "{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"1\",\"_version\":1,\"created\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}

335-- Response {responseStatus = Status {statusCode = 201, statusMessage = "Created"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","74")], responseBody = "{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"1\",\"_version\":1,\"created\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}

321indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName

336indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName

322− -> doc -> DocId -> m Reply

337+ -> IndexDocumentSettings -> doc -> DocId -> m Reply

323indexDocument (IndexName indexName)

338indexDocument (IndexName indexName)

324− (MappingName mappingName) document (DocId docId) =

339+ (MappingName mappingName) cfg document (DocId docId) =

325 bindM2 put url (return body)

340 bindM2 put url (return body)

326− where url = joinPath [indexName, mappingName, docId]

341+ where url = addQuery params <$> joinPath [indexName, mappingName, docId]

342+ params = case idsVersionControl cfg of

343+ NoVersionControl -> []

344+ InternalVersion v -> versionParams v "internal"

345+ ExternalGT (ExternalDocVersion v) -> versionParams v "external_gt"

346+ ExternalGTE (ExternalDocVersion v) -> versionParams v "external_gte"

347+ ForceVersion (ExternalDocVersion v) -> versionParams v "force"

348+ vt = T.pack . show . docVersionNumber

349+ versionParams v t = [("version", vt v), ("version_type", t)]

327 body = Just (encode document)

350 body = Just (encode document)

328

351

329-- | 'deleteDocument' is the primary way to delete a single document.

352-- | 'deleteDocument' is the primary way to delete a single document.

@@ -536,3 +559,17 @@ setURI req URI{..} = do

536 Scheme "https" -> True

559 Scheme "https" -> True

537 _ -> False

560 _ -> False

538 theQueryString = [(k , Just v) | (k, v) <- queryPairs uriQuery]

561 theQueryString = [(k , Just v) | (k, v) <- queryPairs uriQuery]

562+

563+-- | Was there an optimistic concurrency control conflict when

564+-- indexing a document?

565+isVersionConflict :: Reply -> Bool

566+isVersionConflict = statusCheck (== 409)

567+

568+isSuccess :: Reply -> Bool

569+isSuccess = statusCheck (inRange (200, 299))

570+

571+isCreated :: Reply -> Bool

572+isCreated = statusCheck (== 201)

573+

574+statusCheck :: (Int -> Bool) -> Reply -> Bool

575+statusCheck prd = prd . NHTS.statusCode . responseStatus

src/Database/Bloodhound/Types.hs+101−1

@@ -29,6 +29,7 @@

29module Database.Bloodhound.Types

30 ( defaultCache

31 , defaultIndexSettings

32+ , defaultIndexDocumentSettings

32 , mkSort

33 , mkSort

33 , showText

34 , showText

34 , unpackId

35 , unpackId

@@ -41,6 +42,8 @@ module Database.Bloodhound.Types

41 , mkTermsAggregation

42 , mkTermsAggregation

42 , mkTermsScriptAggregation

43 , mkTermsScriptAggregation

43 , mkDateHistogram

44 , mkDateHistogram

45+ , mkDocVersion

46+ , docVersionNumber

44 , toTerms

47 , toTerms

45 , toDateHistogram

48 , toDateHistogram

46 , omitNulls

49 , omitNulls

@@ -56,6 +59,10 @@ module Database.Bloodhound.Types

56 , Server(..)

59 , Server(..)

57 , Reply

60 , Reply

58 , EsResult(..)

61 , EsResult(..)

62+ , DocVersion

63+ , ExternalDocVersion(..)

64+ , VersionControl(..)

65+ , IndexDocumentSettings(..)

59 , Query(..)

66 , Query(..)

60 , Search(..)

67 , Search(..)

61 , SearchResult(..)

68 , SearchResult(..)

@@ -218,10 +225,12 @@ import qualified Data.ByteString.Lazy.Char8 as L

218import Data.List (nub)

225import Data.List (nub)

219import Data.List.NonEmpty (NonEmpty (..), toList)

226import Data.List.NonEmpty (NonEmpty (..), toList)

220import qualified Data.Map.Strict as M

227import qualified Data.Map.Strict as M

228+import Data.Maybe

221import Data.Text (Text)

229import Data.Text (Text)

222import qualified Data.Text as T

230import qualified Data.Text as T

223import Data.Time.Clock (UTCTime)

231import Data.Time.Clock (UTCTime)

224import qualified Data.Vector as V

232import qualified Data.Vector as V

233+import GHC.Enum

225import GHC.Generics (Generic)

234import GHC.Generics (Generic)

226import Network.HTTP.Client

235import Network.HTTP.Client

227import qualified Network.HTTP.Types.Method as NHTM

236import qualified Network.HTTP.Types.Method as NHTM

@@ -371,10 +380,78 @@ data BulkOperation =

371data EsResult a = EsResult { _index :: Text

380data EsResult a = EsResult { _index :: Text

372 , _type :: Text

381 , _type :: Text

373 , _id :: Text

382 , _id :: Text

374− , _version :: Int

383+ , _version :: DocVersion

375 , found :: Maybe Bool

384 , found :: Maybe Bool

376 , _source :: a } deriving (Eq, Show)

385 , _source :: a } deriving (Eq, Show)

377

386

387+{-| 'DocVersion' is an integer version number for a document between 1

388+and 9.2e+18 used for <<https://www.elastic.co/guide/en/elasticsearch/guide/current/optimistic-concurrency-control.html optimistic concurrency control>>.

389+-}

390+newtype DocVersion = DocVersion {

391+ docVersionNumber :: Int

392+ } deriving (Eq, Show, Ord, ToJSON)

393+

394+-- | Smart constructor for in-range doc version

395+mkDocVersion :: Int -> Maybe DocVersion

396+mkDocVersion i

397+ | i >= (docVersionNumber minBound) && i <= (docVersionNumber maxBound) =

398+ Just $ DocVersion i

399+ | otherwise = Nothing

400+

401+

402+{-| 'ExternalDocVersion' is a convenience wrapper if your code uses its

403+own version numbers instead of ones from ES.

404+-}

405+newtype ExternalDocVersion = ExternalDocVersion DocVersion

406+ deriving (Eq, Show, Ord, Bounded, Enum, ToJSON)

407+

408+{-| 'VersionControl' is specified when indexing documents as a

409+optimistic concurrency control.

410+-}

411+data VersionControl = NoVersionControl

412+ -- ^ Don't send a version. This is a pure overwrite.

413+ | InternalVersion DocVersion

414+ -- ^ Use the default ES versioning scheme. Only

415+ -- index the document if the version is the same

416+ -- as the one specified. Only applicable to

417+ -- updates, as you should be getting Version from

418+ -- a search result.

419+ | ExternalGT ExternalDocVersion

420+ -- ^ Use your own version numbering. Only index

421+ -- the document if the version is strictly higher

422+ -- OR the document doesn't exist. The given

423+ -- version will be used as the new version number

424+ -- for the stored document. N.B. All updates must

425+ -- increment this number, meaning there is some

426+ -- global, external ordering of updates.

427+ | ExternalGTE ExternalDocVersion

428+ -- ^ Use your own version numbering. Only index

429+ -- the document if the version is equal or higher

430+ -- than the stored version. Will succeed if there

431+ -- is no existing document. The given version will

432+ -- be used as the new version number for the

433+ -- stored document. Use with care, as this could

434+ -- result in data loss.

435+ | ForceVersion ExternalDocVersion

436+ -- ^ The document will always be indexed and the

437+ -- given version will be the new version. This is

438+ -- typically used for correcting errors. Use with

439+ -- care, as this could result in data loss.

440+ deriving (Show, Eq, Ord)

441+

442+{-| 'IndexDocumentSettings' are special settings supplied when indexing

443+a document. For the best backwards compatiblity when new fields are

444+added, you should probably prefer to start with 'defaultIndexDocumentSettings'

445+-}

446+data IndexDocumentSettings = IndexDocumentSettings {

447+ idsVersionControl :: VersionControl

448+ }

449+

450+{-| Reasonable default settings. Chooses no version control.

451+-}

452+defaultIndexDocumentSettings :: IndexDocumentSettings

453+defaultIndexDocumentSettings = IndexDocumentSettings NoVersionControl

454+

378{-| 'Sort' is a synonym for a list of 'SortSpec's. Sort behavior is order

455{-| 'Sort' is a synonym for a list of 'SortSpec's. Sort behavior is order

379 dependent with later sorts acting as tie-breakers for earlier sorts.

456 dependent with later sorts acting as tie-breakers for earlier sorts.

380-}

457-}

@@ -2173,3 +2250,26 @@ instance FromJSON ShardResult where

2173 v .: "successful" <*>

2250 v .: "successful" <*>

2174 v .: "failed"

2251 v .: "failed"

2175 parseJSON _ = empty

2252 parseJSON _ = empty

2253+

2254+

2255+instance FromJSON DocVersion where

2256+ parseJSON v = do

2257+ i <- parseJSON v

2258+ maybe (fail "DocVersion out of range") return $ mkDocVersion i

2259+

2260+instance Bounded DocVersion where

2261+ minBound = DocVersion 1

2262+ maxBound = DocVersion 9200000000000000000 -- 9.2e+18

2263+

2264+instance Enum DocVersion where

2265+ succ x

2266+ | x /= maxBound = DocVersion (succ $ docVersionNumber x)

2267+ | otherwise = succError "DocVersion"

2268+ pred x

2269+ | x /= minBound = DocVersion (pred $ docVersionNumber x)

2270+ | otherwise = predError "DocVersion"

2271+ toEnum i =

2272+ fromMaybe (error $ show i ++ " out of DocVersion range") $ mkDocVersion i

2273+ fromEnum = docVersionNumber

2274+ enumFrom = boundedEnumFrom

2275+ enumFromThen = boundedEnumFromThen

tests/tests.hs+42−6

@@ -25,7 +25,7 @@ import Network.HTTP.Client

25import qualified Network.HTTP.Types.Status as NHTS

26import Prelude hiding (filter)

27import Test.Hspec

28−import Test.QuickCheck.Property.Monoid

28+import Test.QuickCheck.Property.Monoid (prop_Monoid, eq, T(..))

30import Test.Hspec.QuickCheck (prop)

31import Test.QuickCheck

@@ -147,18 +147,27 @@ otherTweet = Tweet { user = "notmyapp"

147 , age = 1000

148 , location = Location 40.12 (-71.34) }

149

150−insertData :: BH IO ()

150+resetIndex :: BH IO ()

151−insertData = do

151+resetIndex = do

152 _ <- deleteExampleIndex

153 _ <- createExampleIndex

154 _ <- putMapping testIndex testMapping TweetMapping

155− _ <- indexDocument testIndex testMapping exampleTweet (DocId "1")

156− _ <- refreshIndex testIndex

157 return ()

155 return ()

158

156

157+insertData :: BH IO Reply

158+insertData = do

159+ resetIndex

160+ insertData' defaultIndexDocumentSettings

161+

162+insertData' :: IndexDocumentSettings -> BH IO Reply

163+insertData' ids = do

164+ r <- indexDocument testIndex testMapping ids exampleTweet (DocId "1")

165+ _ <- refreshIndex testIndex

166+ return r

167+

159insertOther :: BH IO ()

168insertOther :: BH IO ()

160insertOther = do

169insertOther = do

161− _ <- indexDocument testIndex testMapping otherTweet (DocId "2")

170+ _ <- indexDocument testIndex testMapping defaultIndexDocumentSettings otherTweet (DocId "2")

162 _ <- refreshIndex testIndex

171 _ <- refreshIndex testIndex

163 return ()

172 return ()

164

173

@@ -290,6 +299,14 @@ main = hspec $ do

290 (responseBody docInserted) :: Either String (EsResult Tweet)

299 (responseBody docInserted) :: Either String (EsResult Tweet)

291 liftIO $ (fmap _source newTweet `shouldBe` Right exampleTweet)

300 liftIO $ (fmap _source newTweet `shouldBe` Right exampleTweet)

292

301

302+ it "can use optimistic concurrency control" $ withTestEnv $ do

303+ let ev = ExternalDocVersion minBound

304+ let cfg = defaultIndexDocumentSettings { idsVersionControl = ExternalGT ev }

305+ resetIndex

306+ res <- insertData' cfg

307+ liftIO $ isCreated res `shouldBe` True

308+ res' <- insertData' cfg

309+ liftIO $ isVersionConflict res' `shouldBe` True

293

310

294 describe "bulk API" $ do

311 describe "bulk API" $ do

295 it "inserts all documents we request" $ withTestEnv $ do

312 it "inserts all documents we request" $ withTestEnv $ do

@@ -659,3 +676,22 @@ main = hspec $ do

659 describe "Monoid (SearchHits a)" $ do

676 describe "Monoid (SearchHits a)" $ do

660 prop "abides the monoid laws" $ eq $

677 prop "abides the monoid laws" $ eq $

661 prop_Monoid (T :: T (SearchHits ()))

678 prop_Monoid (T :: T (SearchHits ()))

679+

680+ describe "mkDocVersion" $ do

681+ prop "can never construct an out of range docVersion" $ \i ->

682+ let res = mkDocVersion i

683+ in case res of

684+ Nothing -> property True

685+ Just dv -> (dv >= minBound) .&&.

686+ (dv <= maxBound) .&&.

687+ docVersionNumber dv === i

688+

689+ describe "Enum DocVersion" $ do

690+ it "follows the laws of Enum, Bounded" $ do

691+ return (succ maxBound :: DocVersion) `shouldThrow` anyErrorCall

692+ return (pred minBound :: DocVersion) `shouldThrow` anyErrorCall

693+ return (toEnum 0 :: DocVersion) `shouldThrow` anyErrorCall

694+ return (toEnum 9200000000000000001 :: DocVersion) `shouldThrow` anyErrorCall

695+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]

696+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]

697+ enumFromThen minBound (pred maxBound :: DocVersion) `shouldBe` [minBound, pred maxBound, maxBound]