Repository: dgraph-io/badger
Pull request #1412 (2020-07-10 18:44:14)

Add a limit to the size of the batches sent over a stream.
Currently, there's no limit to how many KVs can be included in a single
batch. This leads to issues when the size is over a hard limit. For
example, a batch of size > 2GB will cause issues if it has to be sent
over gRPC.

<!-- 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/1412)
<!-- Reviewable:end -->



Patch:
diff --git a/.github/ISSUE_TEMPLATE b/.github/ISSUE_TEMPLATE
index 893e1e0..7b0145e 100644
--- a/.github/ISSUE_TEMPLATE
+++ b/.github/ISSUE_TEMPLATE
@@ -1,43 +1,1 @@
-<!-- Please answer these questions before submitting your issue. Thanks! -->
-
-### What version of Go are you using (`go version`)?
-
-<pre>
-$ go version
-
-</pre>
-
-### What operating system are you using?
-
-
-
-### What version of Badger are you using?
-
-
-
-### Does this issue reproduce with the latest master?
-
-
-
-### Steps to Reproduce the issue
-<!--
-If possible, provide a recipe for reproducing the error. A complete runnable program is great.
--->
-
-
-### What Badger options were set? 
-<!--
-Provide the list of Badger options that were set.
--->
-
-
-
-### What did you do?
-
-
-
-### What did you expect to see?
-
-
-
-### What did you see instead?
+**GitHub Issues are deprecated. Use [Discuss Issues](https://discuss.dgraph.io/c/issues/badger/37) for reporting issues about this repository.**
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
new file mode 100644
index 0000000..5f1aa26
--- /dev/null
+++ b/.github/workflows/main.yml
@@ -0,0 +1,24 @@
+# This is a basic workflow to help you get started with Actions
+
+name: Issue Closer
+
+# Controls when the action will run. Triggers the workflow on push or pull request
+# events but only for the master branch
+on:
+  issues:
+    types: [ opened, reopened ]
+
+# A workflow run is made up of one or more jobs that can run sequentially or in parallel
+jobs:
+  # This workflow contains a single job called "build"
+  build:
+    # The type of runner that the job will run on
+    runs-on: ubuntu-latest
+
+    # Steps represent a sequence of tasks that will be executed as part of the job
+    steps:
+      - name: Close Issue
+        uses: peter-evans/close-issue@v1.0.1
+        with:
+          comment: |
+           **Use [Discuss Issues](https://discuss.dgraph.io/c/issues/badger/37) for reporting issues about this repository.**
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1ef2ce9..9f4200b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,8 +1,92 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
-The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
-and this project adheres to [Serialization Versioning](VERSIONING.md).
+The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
+
+## [20.07.0-rc1] - 2020-07-11
+
+### Fixed
+  - Fix Sequence generates duplicate values (#1281)
+  - Fix race condition in DoesNotHave (#1287)
+  - Fail fast if cgo is disabled and compression is ZSTD (#1284)
+  - Proto: make badger/v2 compatible with v1 (#1293)
+  - Proto: Rename dgraph.badger.v2.pb to badgerpb2 (#1314)
+  - Handle duplicates in ManagedWriteBatch (#1315)
+  - Ensure `bitValuePointer` flag is cleared for LSM entry values written to LSM (#1313)
+  - DropPrefix: Return error on blocked writes (#1329)
+  - Confirm `badgerMove` entry required before rewrite (#1302)
+  - Drop move keys when its key prefix is dropped (#1331)
+  - Iterator: Always add key to txn.reads (#1328)
+  - Restore: Account for value size as well (#1358)
+  - Compaction: Expired keys and delete markers are never purged (#1354)
+  - GC: Consider size of value while rewriting (#1357)
+  - Force KeepL0InMemory to be true when InMemory is true (#1375)
+  - Rework DB.DropPrefix (#1381)
+  - Update head while replaying value log (#1372)
+  - Avoid panic on multiple closer.Signal calls (#1401)
+  - Return error if the vlog writes exceeds more than 4GB (#1400)
+
+### Performance
+  - Clean up transaction oracle as we go (#1275)
+  - Use cache for storing block offsets (#1336)
+
+### Features
+  - Support disabling conflict detection (#1344)
+  - Add leveled logging (#1249)
+  - Support entry version in Write batch (#1310)
+  - Add Write method to batch write (#1321)
+  - Support multiple iterators in read-write transactions (#1286)
+
+### New APIs
+- Badger.DB
+  - NewManagedWriteBatch (#1310)
+  - DropPrefix (#1381)
+- Badger.Option
+  - WithDetectConflicts (#1344)
+  - WithKeepBlockIndicesInCache (#1336)
+  - WithKeepBlocksInCache (#1336)
+- Badger.WriteBatch
+  - DeleteAt (#1310)
+  - SetEntryAt (#1310)
+  - Write (#1321)
+
+### Changes to Default Options
+  - DefaultOptions: Set KeepL0InMemory to false (#1345)
+  - Increase default valueThreshold from 32B to 1KB (#1346)
+
+### Deprecated
+- Badger.Option
+  - WithEventLogging (#1203)
+
+### Reverts
+This sections lists the changes which were reverted because of non-reproducible crashes.
+- Compress/Encrypt Blocks in the background (#1227)
+
+
+## [2.0.3] - 2020-03-24
+
+### Fixed
+
+- Add support for watching nil prefix in subscribe API (#1246)
+
+### Performance
+
+- Compress/Encrypt Blocks in the background (#1227)
+- Disable cache by default (#1257)
+
+### Features
+
+- Add BypassDirLock option (#1243)
+- Add separate cache for bloomfilters (#1260)
+
+### New APIs
+- badger.DB
+  - BfCacheMetrics (#1260)
+  - DataCacheMetrics (#1260)
+- badger.Options
+  - WithBypassLockGuard (#1243)
+  - WithLoadBloomsOnOpen (#1260)
+  - WithMaxBfCacheSize (#1260)
 
 ## [2.0.3] - 2020-03-24
 
@@ -318,7 +402,8 @@ Bug fix:
 ## [1.0.1] - 2017-11-06
 * Fix an uint16 overflow when resizing key slice
 
-[Unreleased]: https://github.com/dgraph-io/badger/compare/v2.0.3...HEAD
+[Unreleased]: https://github.com/dgraph-io/badger/compare/v20.07.0-rc1...HEAD
+[20.07.0-rc1]: https://github.com/dgraph-io/badger/compare/v2.0.3...v20.07.0-rc1
 [2.0.3]: https://github.com/dgraph-io/badger/compare/v2.0.2...v2.0.3
 [2.0.2]: https://github.com/dgraph-io/badger/compare/v2.0.1...v2.0.2
 [2.0.1]: https://github.com/dgraph-io/badger/compare/v2.0.0...v2.0.1
diff --git a/README.md b/README.md
index 2a22091..5e00e82 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,8 @@ in pure Go. It is the underlying database for [Dgraph](https://dgraph.io), a
 fast, distributed graph database. It's meant to be a performant alternative to
 non-Go-based key-value stores like RocksDB.
 
+**Use [Discuss Issues](https://discuss.dgraph.io/c/issues/badger/37) for reporting issues about this repository.**
+
 ## Project Status [March 24, 2020]
 
 Badger is stable and is being used to serve data sets worth hundreds of
diff --git a/levels.go b/levels.go
index 71f0a36..ecacdba 100644
--- a/levels.go
+++ b/levels.go
@@ -663,6 +663,9 @@ nextTable:
 			return tbl, errors.Wrapf(err, "Unable to open table: %q", fd.Name())
 		}
 		if builder.Empty() {
+			// Cleanup builder resources:
+			builder.Finish()
+			builder.Close()
 			continue
 		}
 		numBuilds++
diff --git a/stream.go b/stream.go
index 30814e7..eb7341e 100644
--- a/stream.go
+++ b/stream.go
@@ -32,6 +32,12 @@ import (
 
 const pageSize = 4 << 20 // 4MB
 
+// maxStreamSize is the maximum allowed size of a stream batch. This is a soft limit
+// as a single list that is still over the limit will have to be sent as is since it
+// cannot be split further. This limit prevents the framework from creating batches
+// so big that sending them causes issues (e.g running into the max size gRPC limit).
+var maxStreamSize = uint64(100 << 20) // 100MB
+
 // Stream provides a framework to concurrently iterate over a snapshot of Badger, pick up
 // key-values, batch them up and call Send. Stream does concurrent iteration over many smaller key
 // ranges. It does NOT send keys in lexicographical sorted order. To get keys in sorted
@@ -248,9 +254,29 @@ func (st *Stream) streamKVs(ctx context.Context) error {
 	defer t.Stop()
 	now := time.Now()
 
+	sendBatch := func(batch *pb.KVList) error {
+		sz := uint64(proto.Size(batch))
+		bytesSent += sz
+		count += len(batch.Kv)
+		t := time.Now()
+		if err := st.Send(batch); err != nil {
+			return err
+		}
+		st.db.opt.Infof("%s Created batch of size: %s in %s.\n",
+			st.LogPrefix, humanize.Bytes(sz), time.Since(t))
+		return nil
+	}
+
 	slurp := func(batch *pb.KVList) error {
 	loop:
 		for {
+			// Send the batch immediately if it already exceeds the maximum allowed size.
+			// If the size of the batch exceeds maxStreamSize, break from the loop to
+			// avoid creating a batch that is so big that certain limits are reached.
+			sz := uint64(proto.Size(batch))
+			if sz > maxStreamSize {
+				break loop
+			}
 			select {
 			case kvs, ok := <-st.kvChan:
 				if !ok {
@@ -262,16 +288,7 @@ func (st *Stream) streamKVs(ctx context.Context) error {
 				break loop
 			}
 		}
-		sz := uint64(proto.Size(batch))
-		bytesSent += sz
-		count += len(batch.Kv)
-		t := time.Now()
-		if err := st.Send(batch); err != nil {
-			return err
-		}
-		st.db.opt.Infof("%s Created batch of size: %s in %s.\n",
-			st.LogPrefix, humanize.Bytes(sz), time.Since(t))
-		return nil
+		return sendBatch(batch)
 	}
 
 outer:
@@ -297,6 +314,8 @@ outer:
 			}
 			y.AssertTrue(kvs != nil)
 			batch = kvs
+
+			// Otherwise, slurp more keys into this batch.
 			if err := slurp(batch); err != nil {
 				return err
 			}


Test patch:
diff --git a/stream_test.go b/stream_test.go
index f60e3de..f3c4b1d 100644
--- a/stream_test.go
+++ b/stream_test.go
@@ -77,9 +77,7 @@ func TestStream(t *testing.T) {
 	stream := db.NewStreamAt(math.MaxUint64)
 	stream.LogPrefix = "Testing"
 	c := &collector{}
-	stream.Send = func(list *bpb.KVList) error {
-		return c.Send(list)
-	}
+	stream.Send = c.Send
 
 	// Test case 1. Retrieve everything.
 	err = stream.Orchestrate(ctxb)
@@ -186,9 +184,7 @@ func TestStreamWithThreadId(t *testing.T) {
 		return stream.ToList(key, itr)
 	}
 	c := &collector{}
-	stream.Send = func(list *bpb.KVList) error {
-		return c.Send(list)
-	}
+	stream.Send = c.Send
 
 	err = stream.Orchestrate(ctxb)
 	require.NoError(t, err)
@@ -207,3 +203,54 @@ func TestStreamWithThreadId(t *testing.T) {
 	}
 	require.NoError(t, db.Close())
 }
+
+func TestBigStream(t *testing.T) {
+	// Set the maxStreamSize to 1MB for the duration of the test so that the it can use a smaller
+	// dataset than it would otherwise need.
+	originalMaxStreamSize := maxStreamSize
+	maxStreamSize = 1 << 20
+	defer func() {
+		maxStreamSize = originalMaxStreamSize
+	}()
+
+	testSize := int(1e6)
+	dir, err := ioutil.TempDir("", "badger-big-test")
+	require.NoError(t, err)
+	defer removeDir(dir)
+
+	db, err := OpenManaged(DefaultOptions(dir))
+	require.NoError(t, err)
+
+	var count int
+	wb := db.NewWriteBatchAt(5)
+	for _, prefix := range []string{"p0", "p1", "p2"} {
+		for i := 1; i <= testSize; i++ {
+			require.NoError(t, wb.SetEntry(NewEntry(keyWithPrefix(prefix, i), value(i))))
+			count++
+		}
+	}
+	require.NoError(t, wb.Flush())
+
+	stream := db.NewStreamAt(math.MaxUint64)
+	stream.LogPrefix = "Testing"
+	c := &collector{}
+	stream.Send = c.Send
+
+	// Test case 1. Retrieve everything.
+	err = stream.Orchestrate(ctxb)
+	require.NoError(t, err)
+	require.Equal(t, 3*testSize, len(c.kv), "Expected 30000. Got: %d", len(c.kv))
+
+	m := make(map[string]int)
+	for _, kv := range c.kv {
+		prefix, ki := keyToInt(kv.Key)
+		expected := value(ki)
+		require.Equal(t, expected, kv.Value)
+		m[prefix]++
+	}
+	require.Equal(t, 3, len(m))
+	for pred, count := range m {
+		require.Equal(t, testSize, count, "Count mismatch for pred: %s", pred)
+	}
+	require.NoError(t, db.Close())
+}