Repository: dgraph-io/badger
Pull request #360 (2017-12-15 10:44:46)

Add MergeOperator to support merge functionality.
* Add a MergeOperator struct

* Added a MergeFunc type that defines a func type to be provided
  by the user to perform the merge operation

* Added methods to support the merge functionality.

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/dgraph-io/badger/360)
<!-- Reviewable:end -->


deepakjois: Tests are TBD.
manishrjain: 



Review status: 0 of 2 files reviewed at latest revision, 16 unresolved discussions.

---

*[db.go, line 1164 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cBP7qc-PjAj8m6DBT:-L0cBP7qc-PjAj8m6DBU:byfo2tg) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1164)):*
> ```Go
> 	db      *DB
> 	key     []byte
> 	version uint64
> ```

Rename to `skipVersion`; so we can check `if item.Version <= op.skipVersion { skip }`

---

*[db.go, line 1175 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c6vcl8flQd9lfg-VR:-L0c6vcmeKR2C__WxZ7b:bxju97q) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1175)):*
> ```Go
> // contains the logic to perform the ‘merge’ and return an updated value.
> // MergeFunc could perform operations like, integer addition, string concatenation
> // or list appends.
> ```

Also mention somewhere that the order might not be retained.

---

*[db.go, line 1179 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c74I992X6_iyi-8LW:-L0c74I992X6_iyi-8LX:bo8gh5t) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1179)):*
> ```Go
> 
> // GetMergeOperator creates a new MergeOperator for a given key and returns
> // a pointer to it. It also fires off a goroutine that performs a merge operation
> ```

that performs a compaction using the merge function.

---

*[db.go, line 1195 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9x2UjnZ6kYA478ge:-L0c9x2UjnZ6kYA478gf:b8yox5k) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1195)):*
> ```Go
> }
> 
> func (op *MergeOperator) merge() error {
> ```

func can be called compact.

---

*[db.go, line 1206 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9XvMUxj9eT3zPUKD:-L0c9XvMUxj9eT3zPUKE:biob1uq) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1206)):*
> ```Go
> 		for it.Rewind(); it.ValidForPrefix(op.key); it.Next() {
> 			item := it.Item()
> 			y.AssertTrue(item.Version() >= op.version)
> ```

greater than.

---

*[db.go, line 1207 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cBCs9H5BpT_Jpn-Ew:-L0cBCs9H5BpT_Jpn-Ex:b-rzhq6s) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1207)):*
> ```Go
> 			item := it.Item()
> 			y.AssertTrue(item.Version() >= op.version)
> 			if item.Version() == op.version {
> ```

Set max of what we read here, to op.version.

---

*[db.go, line 1216 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9c-j2ZiRvGSW1MaY:-L0c9c-j2ZiRvGSW1MaZ:b2niebg) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1216)):*
> ```Go
> 				// Set a flag if value was updated
> 				updated = true
> 				new, err := item.Value()
> ```

val

---

*[db.go, line 1220 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9d97PlWlo-KWpl9Y:-L0c9d97PlWlo-KWpl9Z:b2niebg) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1220)):*
> ```Go
> 					return err
> 				}
> 				existing = op.f(existing, new)
> ```

val

---

*[db.go, line 1226 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9gxeaM47XC_DKtCz:-L0c9gxeaM47XC_DKtD-:bc17d2z) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1226)):*
> ```Go
> 		// Write value back to db, and update version
> 		if updated {
> 			err := txn.Set(op.key, existing)
> ```

If can be in the same line.

---

*[db.go, line 1234 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9ncm1R7AKiXE4OMb:-L0c9ncm1R7AKiXE4OMc:b-xxff1d) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1234)):*
> ```Go
> 
> 		// Purge all previous entries
> 		return op.db.PurgeVersionsBelow(op.key, op.version)
> ```

This should be done outside the txn.

---

*[db.go, line 1238 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cA0BDXKrljbrNMEuC:-L0cA0BDXKrljbrNMEuD:b-b2pxko) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1238)):*
> ```Go
> }
> 
> func (op *MergeOperator) runMerge() {
> ```

runCompactions

---

*[db.go, line 1254 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cA9HzCEXu_88Pcsaj:-L0cA9HzCEXu_88Pcsak:bk87fo0) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1254)):*
> ```Go
> // routine into the values that were recorded by previous invocations to Add().
> func (op *MergeOperator) Add(val []byte) error {
> 	if op.stopped {
> ```

This isn't being protected by anything. But anyhow, why bother with checking this?

---

*[db.go, line 1288 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAlk1FjDNh-uO_h20:-L0cAlk1FjDNh-uO_h21:bpnvzu7) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1288)):*
> ```Go
> 				}
> 			} else {
> 				new, err := item.Value()
> ```

Keep the naming convention same. In fact, you could break this out into common logic shared by compact and Get.

---

*[db.go, line 1301 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAKZxu9QEm1zg6HG8:-L0cAKZxu9QEm1zg6HG9:b-9xykac) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1301)):*
> ```Go
> 
> // Stop waits for any pending merge to complete and then stops the background
> // goroutine. After Stop() is called, any subsequent calls to Add will result in
> ```

No need to block Add. The users can deal with this at their end.

---

*[db.go, line 1304 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAU1XRoOOfjd9nXc5:-L0cAU1XRoOOfjd9nXc6:b-oy48rg) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1304)):*
> ```Go
> // an error
> func (op *MergeOperator) Stop() error {
> 	op.ticker.Stop()
> ```

Can use y.LevelCloser for this, signal and wait behavior.

---

*[db.go, line 1306 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAYK9DL16Q6bQRspJ:-L0cAYK9DL16Q6bQRspK:bonyh37) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1306)):*
> ```Go
> 	op.ticker.Stop()
> 	err := <-op.done
> 	op.stopped = true
> ```

We don't really need to check for stopped.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

deepakjois: 



Review status: 0 of 4 files reviewed at latest revision, 16 unresolved discussions.

---

*[db.go, line 1164 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cBP7qc-PjAj8m6DBT:-L0cH8RTYLWb3mX_HuWN:b3l3go8) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1164)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Rename to `skipVersion`; so we can check `if item.Version <= op.skipVersion { skip }`
</blockquote></details>

Renamed to `skipVersion`. We don’t really need to check if `item.Version <= op.skipVersion` because they are purged from the DB after they are merged.

---

*[db.go, line 1175 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c6vcl8flQd9lfg-VR:-L0cI9REIK2KxSf9sRyb:b-s1324m) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1175)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Also mention somewhere that the order might not be retained.
</blockquote></details>

Will do. But this does limit the usefulness of this function in some scenarios like string appends, which typically rely on ordering

---

*[db.go, line 1179 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c74I992X6_iyi-8LW:-L0cNFDVfw9LiNSO8mm5:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1179)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

that performs a compaction using the merge function.
</blockquote></details>

Done.

---

*[db.go, line 1195 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9x2UjnZ6kYA478ge:-L0cKIAmwnwtymuHAYRa:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1195)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

func can be called compact.
</blockquote></details>

Done.

---

*[db.go, line 1206 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9XvMUxj9eT3zPUKD:-L0cHSCmg-I2SgQvFIF6:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1206)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

greater than.
</blockquote></details>

Done.

---

*[db.go, line 1207 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cBCs9H5BpT_Jpn-Ew:-L0cHTIXb2UgCq6pmDRi:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1207)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Set max of what we read here, to op.version.
</blockquote></details>

Done.

---

*[db.go, line 1216 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9c-j2ZiRvGSW1MaY:-L0cNJAop-7JCb_HJIFL:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1216)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

val
</blockquote></details>

Done.

---

*[db.go, line 1220 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9d97PlWlo-KWpl9Y:-L0cNK5Sb1MRpVIHkF-K:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1220)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

val
</blockquote></details>

Done.

---

*[db.go, line 1226 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9gxeaM47XC_DKtCz:-L0cNhziq5PXlTN13oN0:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1226)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

If can be in the same line.
</blockquote></details>

Done.

---

*[db.go, line 1234 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0c9ncm1R7AKiXE4OMb:-L0cNN6Jl2NKewb4QqW1:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1234)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

This should be done outside the txn.
</blockquote></details>

Done.

---

*[db.go, line 1238 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cA0BDXKrljbrNMEuC:-L0cOOTQsJOrFxlq4T27:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1238)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

runCompactions
</blockquote></details>

Done.

---

*[db.go, line 1254 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cA9HzCEXu_88Pcsaj:-L0cOLs223mxo0R5SmWb:b-z9wj3m) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1254)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

This isn't being protected by anything. But anyhow, why bother with checking this?
</blockquote></details>

Removed `stopped`

---

*[db.go, line 1288 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAlk1FjDNh-uO_h20:-L0cPS40DuAsWfpSaZio:bfcds8n) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1288)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Keep the naming convention same. In fact, you could break this out into common logic shared by compact and Get.
</blockquote></details>

I looked at doing it, but the logic is slightly different in the `else` block. So it may not be worth refactoring. Maybe the creation of the iterator and the `then` block can be factored out, but the code would look a bit convoluted.

Open to any alternate suggestions.

---

*[db.go, line 1301 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAKZxu9QEm1zg6HG8:-L0cOQqkJxNay-F5jBzD:bmsfpjr) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1301)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

No need to block Add. The users can deal with this at their end.
</blockquote></details>

Done.

Just to be sure, you are saying clients of the API should ensure `Add()` is not called after `Stop()`. The behavior is undefined.

---

*[db.go, line 1304 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAU1XRoOOfjd9nXc5:-L0cZSITt7jA4q1i58su:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1304)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Can use y.LevelCloser for this, signal and wait behavior.
</blockquote></details>

Done.

---

*[db.go, line 1306 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAYK9DL16Q6bQRspJ:-L0cOSYhY_LoWnRJTlJ2:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1306)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

We don't really need to check for stopped.
</blockquote></details>

Done.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

deepakjois: 



Review status: 2 of 4 files reviewed at latest revision, 13 unresolved discussions.

---

*[db.go, line 1288 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cAlk1FjDNh-uO_h20:-L0dMwL9G2i6gsAcu_TQ:b1fg3nt) ([raw file](https://github.com/dgraph-io/badger/blob/5233e662ebe5fb9d8c46214786d0135f38b85b31/db.go#L1288)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

You mean the updated flag? You can have a func which iterates using the itr, and merges the values, returning back the final value and a count of items iterated.

If count > 1, then you have the update. Otherwise, not.
</blockquote></details>

I have added an extra method which is called by both `Get()` and `compact`. It is a bit awkward but its probably okay given we have the logic in one place.

---

*[db.go, line 1183 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cm0iki6wxVQ1Mbdpg:-L0d5f93LyK3oorFE-li:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1183)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Don't need to store the ticker. Closer can take care of exiting the goroutine.
</blockquote></details>

Done.

---

*[db.go, line 1184 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cplRyh2lspBbHUgF8:-L0d5gSH3Biu7X1A0aDL:b-68kswg) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1184)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

No need for err here.
</blockquote></details>

This is to record any errors that happen in the background goroutine. If you see the code for `MergeOperator.Get()`, I check for err there by looking at this field. 

---

*[db.go, line 1226 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0co-n04Wbw5ku8Cntx:-L0d8nni4YSG0owR5jeg:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1226)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Instead of asserting this, just skip.
</blockquote></details>

Done.

---

*[db.go, line 1230 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cmeNldmHrqe6Pg0l4:-L0d6-IuKyDgT1Vs1OCz:b-vipjix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1230)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

// if len(existing) == 0 {

This can do the role of first.
</blockquote></details>

An earlier version of the code did that. But I realised it is possible to insert blank values (empty slice) into the DB.

It would be preferable to not reduce the usefulness of the merge functionality by restricting the use of blank values. So I removed that check in favor of this.

---

*[db.go, line 1253 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cmwkPP1wa3k-R6J-0:-L0d8prhq44JJtNWUmk3:bno0asn) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1253)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

This should only be updated, if txn is successful. Otherwise, data would effectively be "lost".
</blockquote></details>

Done.

Updating it outside the transaction now, once it is successfully committed.

---

*[db.go, line 1261 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cnw3p3L-sxe7FidPH:-L0d9KmUzT7nLEjOp24z:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1261)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

No need to block Gets for this. You can call this from runCompactions, so the lock isn't held for this.
</blockquote></details>

Done.

---

*[db.go, line 1274 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cnKNyqxLS3c8K8_Sy:-L0dANuj5MwO9PJ3t9Ad:boceo6s) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1274)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Don't break out. Just log it or something. Also, call Purge here.
</blockquote></details>

I have made the change you suggested. We basically don’t stop the merge operation if errors are encountered.

But. I prefer capturing the error, unless there are good reasons to not do it.

---

*[db.go, line 1280 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cnQo4IDmpxzGok6LF:-L0dBEKRWdbiNMAB27x5:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1280)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

You can defer it right at the top for safety.
</blockquote></details>

Done.

---

*[db.go, line 1309 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cndu323nNHgWWjm7Z:-L0dBHBJ96E32lOHV64x:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1309)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Let's not enforce this. Purging can be a lazy op done in a goroutine.
</blockquote></details>

Done.

---

*[db.go, line 1310 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cofx0oPMCv5enOAoY:-L0dBIDTbQbzUPF-6YII:bdt1w0q) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1310)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Don't need first var. Use len(existing).
</blockquote></details>

See notes about this above. values can be blank.

---

*[db.go, line 1337 at r2](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0cpd1nLX_ZFEie_NbC:-L0d7IcHSYZkijLuLW7z:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/f9395bee8b3f3192732b1cd8c67e48d3ea11d009/db.go#L1337)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

You don't need to explicitly call Stop here.
</blockquote></details>

Done.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

manishrjain: LGTM. One last round, and then should be good to go.

---

Reviewed 2 of 2 files at r3.
Review status: all files reviewed at latest revision, 6 unresolved discussions, some commit checks broke.

---

*[db.go, line 1215 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dTLrqBrqNqDVy0lSX:-L0dTLrr6s2v02eVz-c5:b-hko9sj) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/db.go#L1215)):*
> ```Go
> 	opt.AllVersions = true
> 	it := txn.NewIterator(opt)
> 	var (
> ```

var first bool. No need for block.

---

*[db.go, line 1304 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dQvuVIw8VnW8TZMMo:-L0dQvuVIw8VnW8TZMMp:b9eoi8z) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/db.go#L1304)):*
> ```Go
> // Add records a value in Badger which will eventually be merged by a background
> // routine into the values that were recorded by previous invocations to Add().
> func (op *MergeOperator) Add(val []byte) error {
> ```

I'd not allow empty val, but your decision.

---

*[errors.go, line 87 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dSGBVjuUEaAXfmMit:-L0dSGBVjuUEaAXfmMiu:b-6nzv6v) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/errors.go#L87)):*
> ```Go
> 	// ErrStoppedMerge is returned when an MergeOperator.Add is attempted after the
> 	// merge operation has been stopped by calling MergeOperator.Stop
> 	ErrStoppedMerge = errors.New("Cannot add. Merge operation has already been stopped")
> ```

Remove this.

---

*[README.md, line 261 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dT9yBYSc8KiDjDWJm:-L0dT9yBYSc8KiDjDWJn:b-pgr3hq) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L261)):*
> ```Markdown
> ```
> 
> ### Associative Merge Operations
> ```

I'd drop the word Associative.

---

*[README.md, line 285 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dSxzX18epJzmhYmPH:-L0dSxzX18epJzmhYmPI:baa0lpj) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L285)):*
> ```Markdown
> func add(existing, new []byte) []byte {
>   return uint64ToBytes(
>       bytesToUint64(existing) +
> ```

The two funcs are going sideways. So, fix the indentation.

---

*[README.md, line 295 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dT3jZZ7bzAoBd8wHD:-L0dT3jZZ7bzAoBd8wHE:b-x4tjqz) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L295)):*
> ```Markdown
> method.
> 
> `MergeOperator.Get()` method can be used to retrieve the latest value of the key
> ```

s/latest/cumulative?

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360#-:-L0dTdUpzYcKC20-DX21:bvgvhdj)*
<!-- Sent from Reviewable.io -->

deepakjois: 



Review status: all files reviewed at latest revision, 6 unresolved discussions, some commit checks broke.

---

*[db.go, line 1215 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dTLrqBrqNqDVy0lSX:-L0dUs_mDJLJSlfxcBMx:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/db.go#L1215)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

var first bool. No need for block.
</blockquote></details>

Done.

---

*[db.go, line 1304 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dQvuVIw8VnW8TZMMo:-L0dUtcXxCZsQ9-j7XHg:b-62kqpn) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/db.go#L1304)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

I'd not allow empty val, but your decision.
</blockquote></details>

Done.

I had to create a new error `ErrInvalidValue` to throw.

---

*[errors.go, line 87 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dSGBVjuUEaAXfmMit:-L0dUzZkT--vP3Effn0u:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/errors.go#L87)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Remove this.
</blockquote></details>

Done.

---

*[README.md, line 261 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dT9yBYSc8KiDjDWJm:-L0dV-QDxo3h_qORVmnN:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L261)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

I'd drop the word Associative.
</blockquote></details>

Done.

---

*[README.md, line 285 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dSxzX18epJzmhYmPH:-L0dV73S4rAQovSw_XK6:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L285)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

The two funcs are going sideways. So, fix the indentation.
</blockquote></details>

Done.

---

*[README.md, line 295 at r3](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dT3jZZ7bzAoBd8wHD:-L0dVDg2QeFCYMbLEHGJ:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/c0745f557d338c969cb506b27491a0601514b876/README.md#L295)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

s/latest/cumulative?
</blockquote></details>

Done.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

manishrjain: 



Reviewed 3 of 3 files at r4.
Review status: all files reviewed at latest revision, 2 unresolved discussions, some commit checks broke.

---

*[db.go, line 1236 at r4](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dZD-VNGUKneG475X-:-L0dZD-W9vwEq9uFKKk7:b1guo5l) ([raw file](https://github.com/dgraph-io/badger/blob/2d2a539ea460c5959da088bd1368a07a33680f50/db.go#L1236)):*
> ```Go
> 		}
> 	}
> 	if len(val) == 0 {
> ```

Looking at this, if `op.f(..)` returns empty val, it could trigger this. So, we can't use this. Actually, yeah, we have to use first, instead of this val.

Given that merge func should be able to return empty val, then one could argue that setting it to empty should also be allowed.

---

*[db.go, line 1302 at r4](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dZZ_BNKmMDP_ef9Dw:-L0dZZ_BNKmMDP_ef9Dx:bsjte78) ([raw file](https://github.com/dgraph-io/badger/blob/2d2a539ea460c5959da088bd1368a07a33680f50/db.go#L1302)):*
> ```Go
> func (op *MergeOperator) Add(val []byte) error {
> 	if len(val) == 0 {
> 		return ErrInvalidValue
> ```

See above comment.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

deepakjois: 



Review status: 2 of 3 files reviewed at latest revision, 2 unresolved discussions, some commit checks broke.

---

*[db.go, line 1236 at r4](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dZD-VNGUKneG475X-:-L0gq4nXvONsG65N88q3:b1g5bbj) ([raw file](https://github.com/dgraph-io/badger/blob/2d2a539ea460c5959da088bd1368a07a33680f50/db.go#L1236)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Looking at this, if `op.f(..)` returns empty val, it could trigger this. So, we can't use this. Actually, yeah, we have to use first, instead of this val.

Given that merge func should be able to return empty val, then one could argue that setting it to empty should also be allowed.
</blockquote></details>

Using a boolean value `first` now.

---

*[db.go, line 1302 at r4](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0dZZ_BNKmMDP_ef9Dw:-L0gpz3z3SbHv0Helpbj:bo93coz) ([raw file](https://github.com/dgraph-io/badger/blob/2d2a539ea460c5959da088bd1368a07a33680f50/db.go#L1302)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

See above comment.
</blockquote></details>

Removed this

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->

manishrjain: <img class="emoji" title=":lgtm:" alt=":lgtm:" align="absmiddle" src="https://reviewable.io/lgtm.png" height="20" width="61"/> One comment.

---

Reviewed 1 of 2 files at r5.
Review status: all files reviewed at latest revision, 2 unresolved discussions, some commit checks broke.

---

*[db.go, line 1182 at r5](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0grrffAh0QInSg1FMa:-L0grrfgLH8DYp9p2gaU:bs9ugvc) ([raw file](https://github.com/dgraph-io/badger/blob/1fe53738445b965c0af527cafc001bb144667934/db.go#L1182)):*
> ```Go
> 	db          *DB
> 	key         []byte
> 	skipVersion uint64
> ```

Looking at this with fresh eyes, and the naming seems wonky still. How about skipAtOrBelow?

---

*[db.go, line 1294 at r5](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0gshI-y1BWbDGmT6th:-L0gshI-y1BWbDGmT6ti:b-mfzqbt) ([raw file](https://github.com/dgraph-io/badger/blob/1fe53738445b965c0af527cafc001bb144667934/db.go#L1294)):*
> ```Go
> 			}
> 		}
> 		if stop { // timer has stopped
> ```

// timer has stopped is incorrect. Just skip the comment.

Also, do stop the ticker, otherwise, it would continue using CPU.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360#-:-L0gt2UUzDweoNPvHqTt:bpjwta0)*
<!-- Sent from Reviewable.io -->

deepakjois: 



Review status: 2 of 3 files reviewed at latest revision, 2 unresolved discussions, some commit checks broke.

---

*[db.go, line 1182 at r5](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0grrffAh0QInSg1FMa:-L0hBimFKp1cyeSTEZKO:brjqa70) ([raw file](https://github.com/dgraph-io/badger/blob/1fe53738445b965c0af527cafc001bb144667934/db.go#L1182)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Looking at this with fresh eyes, and the naming seems wonky still. How about skipAtOrBelow?
</blockquote></details>

Ya, I had though I will make that change and it slipped my mind. Changed name now.

---

*[db.go, line 1294 at r5](https://reviewable.io:443/reviews/dgraph-io/badger/360#-L0gshI-y1BWbDGmT6th:-L0hBYbNYI4zCQ22zRRa:b-sce64i) ([raw file](https://github.com/dgraph-io/badger/blob/1fe53738445b965c0af527cafc001bb144667934/db.go#L1294)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

// timer has stopped is incorrect. Just skip the comment.

Also, do stop the ticker, otherwise, it would continue using CPU.
</blockquote></details>

Done. Calling the ticker.Stop() method

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/360)*
<!-- Sent from Reviewable.io -->


Patch:
diff --git a/README.md b/README.md
index af84d95..f8dc999 100644
--- a/README.md
+++ b/README.md
@@ -30,6 +30,7 @@ version.
       - [Managing transactions manually](#managing-transactions-manually)
     + [Using key/value pairs](#using-keyvalue-pairs)
     + [Monotonically increasing integers](#monotonically-increasing-integers)
+    * [Merge Operations](#merge-operations)
     + [Setting Time To Live(TTL) and User Metadata on Keys](#setting-time-to-livettl-and-user-metadata-on-keys)
     + [Iterating over keys](#iterating-over-keys)
       - [Prefix scans](#prefix-scans)
@@ -257,6 +258,51 @@ for {
 }
 ```
 
+### Merge Operations
+Badger provides support for unordered merge operations. You can define a func
+of type `MergeFunc` which takes in an existing value, and a value to be
+_merged_ with it. It returns a new value which is the result of the _merge_
+operation. All values are specified in byte arrays. For e.g., here is a merge
+function (`add`) which adds a `uint64` value to an existing `uint64` value.
+
+```Go
+uint64ToBytes(i uint64) []byte {
+  var buf [8]byte
+  binary.BigEndian.PutUint64(buf[:], i)
+  return buf[:]
+}
+
+func bytesToUint64(b []byte) uint64 {
+  return binary.BigEndian.Uint64(b)
+}
+
+// Merge function to add two uint64 numbers
+func add(existing, new []byte) []byte {
+  return uint64ToBytes(bytesToUint64(existing) + bytesToUint64(new))
+}
+```
+
+This function can then be passed to the `DB.GetMergeOperator()` method, along
+with a key, and a duration value. The duration specifies how often the merge
+function is run on values that have been added using the `MergeOperator.Add()`
+method.
+
+`MergeOperator.Get()` method can be used to retrieve the cumulative value of the key
+associated with the merge operation.
+
+```Go
+key := []byte("merge")
+m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+defer m.Stop()
+
+m.Add(uint64ToBytes(1))
+m.Add(uint64ToBytes(2))
+m.Add(uint64ToBytes(3))
+
+res, err := m.Get() // res should have value 6 encoded
+fmt.Println(bytesToUint64(res))
+```
+
 ### Setting Time To Live(TTL) and User Metadata on Keys
 Badger allows setting an optional Time to Live (TTL) value on keys. Once the TTL has
 elapsed, the key will no longer be retrievable and will be eligible for garbage
diff --git a/db.go b/db.go
index 9eddae6..14d50f7 100644
--- a/db.go
+++ b/db.go
@@ -1172,3 +1172,157 @@ func (db *DB) GetSequence(key []byte, bandwidth uint64) (*Sequence, error) {
 	err := seq.updateLease()
 	return seq, err
 }
+
+// MergeOperator represents a Badger merge operator.
+type MergeOperator struct {
+	sync.RWMutex
+	f             MergeFunc
+	db            *DB
+	key           []byte
+	skipAtOrBelow uint64
+	closer        *y.Closer
+}
+
+// MergeFunc accepts two byte slices, one representing an existing value, and
+// another representing a new value that needs to be ‘merged’ into it. MergeFunc
+// contains the logic to perform the ‘merge’ and return an updated value.
+// MergeFunc could perform operations like integer addition, list appends etc.
+// Note that the ordering of the operands is unspecified, so the merge func
+// should either be agnostic to ordering or do additional handling if ordering
+// is required.
+type MergeFunc func(existing, val []byte) []byte
+
+// GetMergeOperator creates a new MergeOperator for a given key and returns a
+// pointer to it. It also fires off a goroutine that performs a compaction using
+// the merge function that runs periodically, as specified by dur.
+func (db *DB) GetMergeOperator(key []byte,
+	f MergeFunc, dur time.Duration) *MergeOperator {
+	op := &MergeOperator{
+		f:      f,
+		db:     db,
+		key:    key,
+		closer: y.NewCloser(1),
+	}
+
+	go op.runCompactions(dur)
+	return op
+}
+
+func (op *MergeOperator) iterateAndMerge(txn *Txn) (maxVersion uint64, val []byte, err error) {
+	opt := DefaultIteratorOptions
+	opt.AllVersions = true
+	it := txn.NewIterator(opt)
+	var first bool
+	for it.Rewind(); it.ValidForPrefix(op.key); it.Next() {
+		item := it.Item()
+		if item.Version() <= op.skipAtOrBelow {
+			continue
+		}
+		if item.Version() > maxVersion {
+			maxVersion = item.Version()
+		}
+		if !first {
+			first = true
+			val, err = item.ValueCopy(val)
+			if err != nil {
+				return 0, nil, err
+			}
+		} else {
+			newVal, err := item.Value()
+			if err != nil {
+				return 0, nil, err
+			}
+			val = op.f(val, newVal)
+		}
+	}
+	if !first {
+		return 0, nil, ErrKeyNotFound
+	}
+	return maxVersion, val, nil
+}
+
+func (op *MergeOperator) compact() error {
+	op.Lock()
+	defer op.Unlock()
+	var maxVersion uint64
+	err := op.db.Update(func(txn *Txn) error {
+		var (
+			val []byte
+			err error
+		)
+		maxVersion, val, err = op.iterateAndMerge(txn)
+		if err != nil {
+			return err
+		}
+
+		// Write value back to db
+		if maxVersion > op.skipAtOrBelow {
+			if err := txn.Set(op.key, val); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil && err != ErrKeyNotFound { // Ignore ErrKeyNotFound errors during compaction
+		return err
+	}
+	// Update version
+	op.skipAtOrBelow = maxVersion
+	return nil
+}
+
+func (op *MergeOperator) runCompactions(dur time.Duration) {
+	ticker := time.NewTicker(dur)
+	defer op.closer.Done()
+	var stop bool
+	for {
+		select {
+		case <-op.closer.HasBeenClosed():
+			stop = true
+		case <-ticker.C: // wait for tick
+		}
+		oldSkipVersion := op.skipAtOrBelow
+		if err := op.compact(); err != nil {
+			log.Printf("Error while running merge operation: %s", err)
+		}
+		// Purge older versions if version has updated
+		if op.skipAtOrBelow > oldSkipVersion {
+			if err := op.db.PurgeVersionsBelow(op.key, op.skipAtOrBelow+1); err != nil {
+				log.Printf("Error purging merged keys: %s", err)
+			}
+		}
+		if stop {
+			ticker.Stop()
+			break
+		}
+	}
+}
+
+// Add records a value in Badger which will eventually be merged by a background
+// routine into the values that were recorded by previous invocations to Add().
+func (op *MergeOperator) Add(val []byte) error {
+	return op.db.Update(func(txn *Txn) error {
+		return txn.Set(op.key, val)
+	})
+}
+
+// Get returns the latest value for the merge operator, which is derived by
+// applying the merge function to all the values added so far.
+//
+// If Add has not been called even once, Get will return ErrKeyNotFound
+func (op *MergeOperator) Get() ([]byte, error) {
+	op.RLock()
+	defer op.RUnlock()
+	var existing []byte
+	err := op.db.View(func(txn *Txn) (err error) {
+		_, existing, err = op.iterateAndMerge(txn)
+		return err
+	})
+	return existing, err
+}
+
+// Stop waits for any pending merge to complete and then stops the background
+// goroutine.
+func (op *MergeOperator) Stop() {
+	op.closer.SignalAndWait()
+}


Test patch:
diff --git a/db_test.go b/db_test.go
index 04827e0..f99878e 100644
--- a/db_test.go
+++ b/db_test.go
@@ -1243,6 +1243,102 @@ func TestSequence_Release(t *testing.T) {
 	})
 }
 
+func uint64ToBytes(i uint64) []byte {
+	var buf [8]byte
+	binary.BigEndian.PutUint64(buf[:], i)
+	return buf[:]
+}
+
+func bytesToUint64(b []byte) uint64 {
+	return binary.BigEndian.Uint64(b)
+}
+
+// Merge function to add two uint64 numbers
+func add(existing, new []byte) []byte {
+	return uint64ToBytes(
+		bytesToUint64(existing) +
+			bytesToUint64(new))
+}
+
+func TestMergeOperatorGetBeforeAdd(t *testing.T) {
+	key := []byte("merge")
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+		defer m.Stop()
+
+		_, err := m.Get()
+		require.Error(t, ErrKeyNotFound, err)
+	})
+}
+
+func TestMergeOperatorBeforeAdd(t *testing.T) {
+	key := []byte("merge")
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+		defer m.Stop()
+		time.Sleep(time.Second)
+	})
+}
+
+func TestMergeOperatorAddAndGet(t *testing.T) {
+	key := []byte("merge")
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+		defer m.Stop()
+
+		err := m.Add(uint64ToBytes(1))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(2))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(3))
+		require.NoError(t, err)
+
+		res, err := m.Get()
+		require.NoError(t, err)
+		require.Equal(t, uint64(6), bytesToUint64(res))
+	})
+}
+
+func TestMergeOperatorCompactBeforeGet(t *testing.T) {
+	key := []byte("merge")
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+		defer m.Stop()
+
+		err := m.Add(uint64ToBytes(1))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(2))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(3))
+		require.NoError(t, err)
+
+		time.Sleep(250 * time.Millisecond) // wait for merge to happen
+
+		res, err := m.Get()
+		require.NoError(t, err)
+		require.Equal(t, uint64(6), bytesToUint64(res))
+	})
+}
+
+func TestMergeOperatorGetAfterStop(t *testing.T) {
+	key := []byte("merge")
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		m := db.GetMergeOperator(key, add, 1*time.Second)
+
+		err := m.Add(uint64ToBytes(1))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(2))
+		require.NoError(t, err)
+		m.Add(uint64ToBytes(3))
+		require.NoError(t, err)
+
+		m.Stop()
+		res, err := m.Get()
+		require.NoError(t, err)
+		require.Equal(t, uint64(6), bytesToUint64(res))
+	})
+}
+
 func ExampleOpen() {
 	dir, err := ioutil.TempDir("", "badger")
 	if err != nil {