opt(stream): add option to directly copy over tables from lower levels (#1700)

Also takes a bug fix from PR #1712, commit 5a8d344

This PR adds FullCopy option in Stream. This allows sending the
table entirely to the writer. If this option is set to true we
directly copy over the tables from the last 2 levels. This option
increases the stream speed while also lowering the memory
consumption on the DB that is streaming the KVs.

For 71GB, compressed and encrypted DB we observed 3x improvement
in speed. The DB contained ~65GB in the last 2 levels while
remaining in the above levels.

To use this option, the following options should be set in Stream.

stream.KeyToList = nil
stream.ChooseKey = nil
stream.SinceTs = 0
db.managedTxns = true

If we use stream writer for receiving the KVs, the encryption mode
has to be the same in sender and receiver. This will restrict
db.StreamDB() to use the same encryption mode in both input and
output DB. Added TODO for allowing different encryption modes.

.golangci.yml

@@ -0,0 +1,47 @@
+version: "2"
+linters:
+  default: none
+  enable:
+    - errcheck
+    - gosec
+    - govet
+    - ineffassign
+    - lll
+    - staticcheck
+    - unconvert
+    - unused
+  settings:
+    gosec:
+      excludes:
+        - G114
+        - G204
+        - G306
+        - G404
+        - G115
+    lll:
+      line-length: 120
+    staticcheck:
+      checks:
+        - all
+        - -SA1019
+  exclusions:
+    generated: lax
+    presets:
+      - comments
+      - common-false-positives
+      - legacy
+      - std-error-handling
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

.trunk/trunk.yaml

@@ -3,20 +3,20 @@
 version: 0.1
 
 cli:
-  version: 1.22.10
+  version: 1.24.0
 
 # Trunk provides extensibility via plugins. (https://docs.trunk.io/plugins)
 plugins:
   sources:
     - id: trunk
-      ref: v1.6.7
+      ref: v1.7.1
       uri: https://github.com/trunk-io/plugins
 
 # Many linters and tools depend on runtimes - configure them here. (https://docs.trunk.io/runtimes)
 runtimes:
   enabled:
     - go@1.23.5
-    - node@18.20.5
+    - node@22.16.0
     - python@3.10.8
 
 # This is the section where you manage your linters. (https://docs.trunk.io/check/configuration)
@@ -26,23 +26,23 @@ lint:
       paths:
         - "*.pb.go"
   enabled:
-    - trivy@0.59.1
-    - renovate@39.169.3
+    - golangci-lint2@2.1.6
+    - trivy@0.63.0
+    - renovate@41.14.0
     - actionlint@1.7.7
-    - checkov@3.2.369
+    - checkov@3.2.447
     - git-diff-check
     - gofmt@1.20.4
-    - golangci-lint@1.63.4
-    - markdownlint@0.44.0
-    - osv-scanner@1.9.2
-    - oxipng@9.1.3
-    - prettier@3.5.0
+    - markdownlint@0.45.0
+    - osv-scanner@2.0.3
+    - oxipng@9.1.5
+    - prettier@3.6.2
     - shellcheck@0.10.0
     - shfmt@3.6.0
-    - svgo@3.3.2
+    - svgo@4.0.0
     - taplo@0.9.3
-    - trufflehog@3.88.7
-    - yamllint@1.35.1
+    - trufflehog@3.89.2
+    - yamllint@1.37.1
 actions:
   enabled:
     - trunk-announce

db.go

@@ -1993,6 +1993,7 @@ func (db *DB) StreamDB(outOptions Options) error {
 	// Stream contents of DB to the output DB.
 	stream := db.NewStreamAt(math.MaxUint64)
 	stream.LogPrefix = fmt.Sprintf("Streaming DB to new DB at %s", outDir)
+	stream.FullCopy = true
 
 	stream.Send = func(buf *z.Buffer) error {
 		return writer.Write(buf)

iterator.go

@@ -355,17 +355,17 @@ func (opt *IteratorOptions) pickTable(t table.TableInterface) bool {
 // that the tables are sorted in the right order.
 func (opt *IteratorOptions) pickTables(all []*table.Table) []*table.Table {
 	filterTables := func(tables []*table.Table) []*table.Table {
-		if opt.SinceTs > 0 {
-			tmp := tables[:0]
-			for _, t := range tables {
-				if t.MaxVersion() < opt.SinceTs {
-					continue
-				}
-				tmp = append(tmp, t)
-			}
-			tables = tmp
+		if opt.SinceTs == 0 {
+			return tables
 		}
-		return tables
+		out := tables[:0]
+		for _, t := range tables {
+			if t.MaxVersion() < opt.SinceTs {
+				continue
+			}
+			out = append(out, t)
+		}
+		return out
 	}
 
 	if len(opt.Prefix) == 0 {
@@ -480,7 +480,7 @@ func (txn *Txn) NewIterator(opt IteratorOptions) *Iterator {
 	for i := 0; i < len(tables); i++ {
 		iters = append(iters, tables[i].sl.NewUniIterator(opt.Reverse))
 	}
-	iters = txn.db.lc.appendIterators(iters, &opt) // This will increment references.
+	iters = append(iters, txn.db.lc.iterators(&opt)...) // This will increment references.
 	res := &Iterator{
 		txn:    txn,
 		iitr:   table.NewMergeIterator(iters, opt.Reverse),

key_registry.go

@@ -10,6 +10,7 @@ import (
 	"crypto/aes"
 	"crypto/rand"
 	"encoding/binary"
+	"errors"
 	"hash/crc32"
 	"io"
 	"os"
@@ -328,8 +329,7 @@ func (kr *KeyRegistry) LatestDataKey() (*pb.DataKey, error) {
 	defer kr.Unlock()
 	// Key might have generated by another go routine. So,
 	// checking once again.
-	key, valid = validKey()
-	if valid {
+	if key, valid := validKey(); valid {
 		return key, nil
 	}
 	k := make([]byte, len(kr.opt.EncryptionKey))
@@ -337,8 +337,8 @@ func (kr *KeyRegistry) LatestDataKey() (*pb.DataKey, error) {
 	if err != nil {
 		return nil, err
 	}
-	_, err = rand.Read(k)
-	if err != nil {
+
+	if _, err := rand.Read(k); err != nil {
 		return nil, err
 	}
 	// Otherwise Increment the KeyID and generate new datakey.
@@ -349,25 +349,40 @@ func (kr *KeyRegistry) LatestDataKey() (*pb.DataKey, error) {
 		CreatedAt: time.Now().Unix(),
 		Iv:        iv,
 	}
-	// Don't store the datakey on file if badger is running in InMemory mode.
-	if !kr.opt.InMemory {
-		// Store the datekey.
-		buf := &bytes.Buffer{}
-		if err = storeDataKey(buf, kr.opt.EncryptionKey, dk); err != nil {
-			return nil, err
-		}
-		// Persist the datakey to the disk
-		if _, err = kr.fp.Write(buf.Bytes()); err != nil {
-			return nil, err
-		}
-	}
-	// storeDatakey encrypts the datakey So, placing un-encrypted key in the memory.
-	dk.Data = k
 	kr.lastCreated = dk.CreatedAt
 	kr.dataKeys[kr.nextKeyID] = dk
+	// Don't store the datakey on file if badger is running in InMemory mode.
+	if kr.opt.InMemory {
+		return dk, nil
+
+	}
+	// Store the datekey.
+	if err = storeDataKey(kr.fp, kr.opt.EncryptionKey, dk); err != nil {
+		return nil, err
+	}
 	return dk, nil
 }
 
+func (kr *KeyRegistry) AddKey(dk *pb.DataKey) (uint64, error) {
+	// If we don't have a encryption key, we cannot store the datakey.
+	if len(kr.opt.EncryptionKey) == 0 {
+		return 0, errors.New("No encryption key found. Cannot add data key")
+	}
+
+	if _, ok := kr.dataKeys[dk.KeyId]; !ok {
+		// If KeyId does not exists already, then use the next available KeyId to store data key.
+		kr.nextKeyID++
+		dk.KeyId = kr.nextKeyID
+	}
+	kr.dataKeys[dk.KeyId] = dk
+
+	if kr.opt.InMemory {
+		return dk.KeyId, nil
+	}
+	// Store the datakey.
+	return dk.KeyId, storeDataKey(kr.fp, kr.opt.EncryptionKey, dk)
+}
+
 // Close closes the key registry.
 func (kr *KeyRegistry) Close() error {
 	if !(kr.opt.ReadOnly || kr.opt.InMemory) {
@@ -377,7 +392,8 @@ func (kr *KeyRegistry) Close() error {
 }
 
 // storeDataKey stores datakey in an encrypted format in the given buffer. If storage key preset.
-func storeDataKey(buf *bytes.Buffer, storageKey []byte, k *pb.DataKey) error {
+// DO NOT use a pointer for key. storeDataKey modifies the kv.Data field.
+func storeDataKey(w io.Writer, storageKey []byte, key *pb.DataKey) error {
 	// xor will encrypt the IV and xor with the given data.
 	// It'll used for both encryption and decryption.
 	xor := func() error {
@@ -385,30 +401,24 @@ func storeDataKey(buf *bytes.Buffer, storageKey []byte, k *pb.DataKey) error {
 			return nil
 		}
 		var err error
-		k.Data, err = y.XORBlockAllocate(k.Data, storageKey, k.Iv)
+		key.Data, err = y.XORBlockAllocate(key.Data, storageKey, key.Iv)
 		return err
 	}
+
 	// In memory datakey will be plain text so encrypting before storing to the disk.
-	var err error
-	if err = xor(); err != nil {
+	if err := xor(); err != nil {
 		return y.Wrapf(err, "Error while encrypting datakey in storeDataKey")
 	}
-	var data []byte
-	if data, err = proto.Marshal(k); err != nil {
-		err = y.Wrapf(err, "Error while marshaling datakey in storeDataKey")
-		var err2 error
-		// decrypting the datakey back.
-		if err2 = xor(); err2 != nil {
-			return y.Wrapf(err,
-				y.Wrapf(err2, "Error while decrypting datakey in storeDataKey").Error())
-		}
-		return err
+
+	data, err := proto.Marshal(key)
+	if err != nil {
+		return y.Wrapf(err, "Error while marshaling datakey in storeDataKey")
 	}
+
 	var lenCrcBuf [8]byte
 	binary.BigEndian.PutUint32(lenCrcBuf[0:4], uint32(len(data)))
 	binary.BigEndian.PutUint32(lenCrcBuf[4:8], crc32.Checksum(data, y.CastagnoliCrcTable))
-	y.Check2(buf.Write(lenCrcBuf[:]))
-	y.Check2(buf.Write(data))
-	// Decrypting the datakey back since we're using the pointer.
-	return xor()
+	y.Check2(w.Write(lenCrcBuf[:]))
+	y.Check2(w.Write(data))
+	return nil
 }

level_handler.go

@@ -293,9 +293,9 @@ func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
 	return maxVs, decr()
 }
 
-// appendIterators appends iterators to an array of iterators, for merging.
+// iterators returns an array of iterators, for merging.
 // Note: This obtains references for the table handlers. Remember to close these iterators.
-func (s *levelHandler) appendIterators(iters []y.Iterator, opt *IteratorOptions) []y.Iterator {
+func (s *levelHandler) iterators(opt *IteratorOptions) []y.Iterator {
 	s.RLock()
 	defer s.RUnlock()
 
@@ -313,14 +313,40 @@ func (s *levelHandler) appendIterators(iters []y.Iterator, opt *IteratorOptions)
 				out = append(out, t)
 			}
 		}
-		return appendIteratorsReversed(iters, out, topt)
+		return iteratorsReversed(out, topt)
 	}
 
 	tables := opt.pickTables(s.tables)
 	if len(tables) == 0 {
-		return iters
+		return nil
 	}
-	return append(iters, table.NewConcatIterator(tables, topt))
+	return []y.Iterator{table.NewConcatIterator(tables, topt)}
+}
+
+func (s *levelHandler) getTables(opt *IteratorOptions) []*table.Table {
+	if opt.Reverse {
+		panic("Invalid option for getTables")
+	}
+
+	s.RLock()
+	defer s.RUnlock()
+
+	if s.level == 0 {
+		var out []*table.Table
+		for _, t := range s.tables {
+			if opt.pickTable(t) {
+				t.IncrRef()
+				out = append(out, t)
+			}
+		}
+		return out
+	}
+
+	tables := opt.pickTables(s.tables)
+	for _, t := range tables {
+		t.IncrRef()
+	}
+	return tables
 }
 
 type levelHandlerRLocked struct{}

levels.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 
+	"github.com/dgraph-io/badger/v4/options"
 	"github.com/dgraph-io/badger/v4/pb"
 	"github.com/dgraph-io/badger/v4/table"
 	"github.com/dgraph-io/badger/v4/y"
@@ -899,7 +900,7 @@ func (s *levelsController) compactBuildTables(
 		var iters []y.Iterator
 		switch {
 		case lev == 0:
-			iters = appendIteratorsReversed(iters, topTables, table.NOCACHE)
+			iters = append(iters, iteratorsReversed(topTables, table.NOCACHE)...)
 		case len(topTables) > 0:
 			y.AssertTrue(len(topTables) == 1)
 			iters = []y.Iterator{topTables[0].NewIterator(table.NOCACHE)}
@@ -1633,7 +1634,8 @@ func (s *levelsController) get(key []byte, maxVs y.ValueStruct, startLevel int)
 	return maxVs, nil
 }
 
-func appendIteratorsReversed(out []y.Iterator, th []*table.Table, opt int) []y.Iterator {
+func iteratorsReversed(th []*table.Table, opt int) []y.Iterator {
+	out := make([]y.Iterator, 0, len(th))
 	for i := len(th) - 1; i >= 0; i-- {
 		// This will increment the reference of the table handler.
 		out = append(out, th[i].NewIterator(opt))
@@ -1641,16 +1643,25 @@ func appendIteratorsReversed(out []y.Iterator, th []*table.Table, opt int) []y.I
 	return out
 }
 
-// appendIterators appends iterators to an array of iterators, for merging.
+// getTables return tables from all levels. It would call IncrRef on all returned tables.
+func (s *levelsController) getTables(opt *IteratorOptions) [][]*table.Table {
+	res := make([][]*table.Table, 0, len(s.levels))
+	for _, level := range s.levels {
+		res = append(res, level.getTables(opt))
+	}
+	return res
+}
+
+// iterators returns an array of iterators, for merging.
 // Note: This obtains references for the table handlers. Remember to close these iterators.
-func (s *levelsController) appendIterators(
-	iters []y.Iterator, opt *IteratorOptions) []y.Iterator {
+func (s *levelsController) iterators(opt *IteratorOptions) []y.Iterator {
 	// Just like with get, it's important we iterate the levels from 0 on upward, to avoid missing
 	// data when there's a compaction.
+	itrs := make([]y.Iterator, 0, len(s.levels))
 	for _, level := range s.levels {
-		iters = level.appendIterators(iters, opt)
+		itrs = append(itrs, level.iterators(opt)...)
 	}
-	return iters
+	return itrs
 }
 
 // TableInfo represents the information about a table.
@@ -1777,3 +1788,46 @@ func (s *levelsController) keySplits(numPerTable int, prefix []byte) []string {
 	sort.Strings(splits)
 	return splits
 }
+
+// AddTable builds the table from the KV.value options passed through the KV.Key.
+func (lc *levelsController) AddTable(
+	kv *pb.KV, lev int, dk *pb.DataKey, change *pb.ManifestChange) error {
+	// TODO: Encryption / Decryption might be required for the table, if the sender and receiver
+	// don't have same encryption mode. See if inplace encryption/decryption can be done.
+	// Tables are sent in the sorted order, so no need to sort them here.
+	encrypted := len(lc.kv.opt.EncryptionKey) > 0
+	y.AssertTrue((dk != nil && encrypted) || (dk == nil && !encrypted))
+	// The keyId is zero if there is no encryption.
+	opts := buildTableOptions(lc.kv)
+	opts.Compression = options.CompressionType(change.Compression)
+	opts.DataKey = dk
+
+	fileID := lc.reserveFileID()
+	fname := table.NewFilename(fileID, lc.kv.opt.Dir)
+
+	// kv.Value is owned by the z.buffer. Ensure that we copy this buffer.
+	var tbl *table.Table
+	var err error
+	if lc.kv.opt.InMemory {
+		if tbl, err = table.OpenInMemoryTable(y.Copy(kv.Value), fileID, &opts); err != nil {
+			return fmt.Errorf("while creating in-memory table from buffer: %w", err)
+		}
+	} else {
+		if tbl, err = table.CreateTableFromBuffer(fname, kv.Value, opts); err != nil {
+			return fmt.Errorf("while creating table from buffer: %w", err)
+		}
+	}
+
+	lc.levels[lev].addTable(tbl)
+	// Release the ref held by OpenTable. addTable would add a reference.
+	_ = tbl.DecrRef()
+
+	change.Id = fileID
+	change.Level = uint32(lev)
+	if dk != nil {
+		change.KeyId = dk.KeyId
+	}
+	// We use the same data KeyId. So, change.KeyId remains the same.
+	y.AssertTrue(change.Op == pb.ManifestChange_CREATE)
+	return lc.kv.manifest.addChanges([]*pb.ManifestChange{change}, lc.kv.opt)
+}

levels_test.go

@@ -7,6 +7,7 @@ package badger
 
 import (
 	"fmt"
+	"io/ioutil"
 	"math"
 	"math/rand"
 	"os"
@@ -29,7 +30,15 @@ func createAndOpen(db *DB, td []keyValVersion, level int) {
 		BloomFalsePositive: db.opt.BloomFalsePositive,
 		ChkMode:            options.NoVerification,
 	}
-	b := table.NewTableBuilder(opts)
+	createAndOpenWithOptions(db, td, level, &opts)
+}
+
+func createAndOpenWithOptions(db *DB, td []keyValVersion, level int, opts *table.Options) {
+	if opts == nil {
+		bopts := buildTableOptions(db)
+		opts = &bopts
+	}
+	b := table.NewTableBuilder(*opts)
 	defer b.Close()
 
 	// Add all keys and versions to the table.
@@ -38,13 +47,21 @@ func createAndOpen(db *DB, td []keyValVersion, level int) {
 		val := y.ValueStruct{Value: []byte(item.val), Meta: item.meta}
 		b.Add(key, val, 0)
 	}
-	fname := table.NewFilename(db.lc.reserveFileID(), db.opt.Dir)
-	tab, err := table.CreateTable(fname, b)
+	fileID := db.lc.reserveFileID()
+	var tab *table.Table
+	var err error
+	if db.opt.InMemory {
+		data := b.Finish()
+		tab, err = table.OpenInMemoryTable(data, fileID, opts)
+	} else {
+		fname := table.NewFilename(fileID, db.opt.Dir)
+		tab, err = table.CreateTable(fname, b)
+	}
 	if err != nil {
 		panic(err)
 	}
 	if err := db.manifest.addChanges([]*pb.ManifestChange{
-		newCreateChange(tab.ID(), level, 0, tab.CompressionType()),
+		newCreateChange(tab.ID(), level, tab.KeyID(), tab.CompressionType()),
 	}, db.opt); err != nil {
 		panic(err)
 	}
@@ -1290,3 +1307,85 @@ func TestStaleDataCleanup(t *testing.T) {
 
 	})
 }
+
+func TestStreamWithFullCopy(t *testing.T) {
+	dbopts := DefaultOptions("")
+	dbopts.managedTxns = true
+	dbopts.MaxLevels = 7
+	dbopts.NumVersionsToKeep = math.MaxInt32
+
+	encKey := make([]byte, 24)
+	_, err := rand.Read(encKey)
+	require.NoError(t, err)
+
+	test := func(db *DB, outOpts Options) {
+		l4 := []keyValVersion{{"a", "1", 3, bitDelete}, {"d", "4", 3, 0}}
+		l5 := []keyValVersion{{"b", "2", 2, 0}}
+		l6 := []keyValVersion{{"a", "1", 2, 0}, {"c", "3", 1, 0}}
+		createAndOpenWithOptions(db, l4, 4, nil)
+		createAndOpenWithOptions(db, l5, 5, nil)
+		createAndOpenWithOptions(db, l6, 6, nil)
+
+		if !outOpts.InMemory {
+			dir, err := ioutil.TempDir("", "badger-test")
+			require.NoError(t, err)
+			defer removeDir(dir)
+			outOpts.Dir = dir
+			outOpts.ValueDir = dir
+		}
+
+		require.NoError(t, db.StreamDB(outOpts))
+		out, err := Open(outOpts)
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, out.Close())
+		}()
+		err = out.View(func(txn *Txn) error {
+			// Key "a" should not be there because we deleted it at higher version.
+			_, err := txn.Get([]byte("a"))
+			require.Error(t, err)
+			require.Equal(t, err, ErrKeyNotFound)
+			_, err = txn.Get([]byte("b"))
+			require.NoError(t, err)
+			_, err = txn.Get([]byte("c"))
+			require.NoError(t, err)
+			_, err = txn.Get([]byte("d"))
+			require.NoError(t, err)
+			return nil
+		})
+		require.NoError(t, err)
+	}
+	t.Run("without encryption", func(t *testing.T) {
+		opts := dbopts
+		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
+			test(db, opts)
+		})
+	})
+	t.Run("with encryption", func(t *testing.T) {
+		opts := dbopts
+		opts.IndexCacheSize = 1 << 20
+		opts.BlockCacheSize = 1 << 20
+		// Set it to zero so that we have more than one data keys.
+		opts.EncryptionKey = encKey
+		opts.EncryptionKeyRotationDuration = 0
+		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
+			test(db, opts)
+			require.Greater(t, len(db.registry.dataKeys), 1)
+		})
+	})
+	t.Run("stream from in-memory to persistent", func(t *testing.T) {
+		opts := dbopts
+		opts.IndexCacheSize = 1 << 20
+		opts.BlockCacheSize = 1 << 20
+		opts.InMemory = true
+		// Set it to zero so that we have more than one data keys.
+		opts.EncryptionKey = encKey
+		opts.EncryptionKeyRotationDuration = 0
+		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
+			outOpts := opts
+			outOpts.InMemory = false
+			test(db, outOpts)
+			require.Greater(t, len(db.registry.dataKeys), 1)
+		})
+	})
+}

manifest.go

@@ -117,7 +117,7 @@ func (m *Manifest) clone(opt Options) Manifest {
 func openOrCreateManifestFile(opt Options) (
 	ret *manifestFile, result Manifest, err error) {
 	if opt.InMemory {
-		return &manifestFile{inMemory: true}, Manifest{}, nil
+		return &manifestFile{inMemory: true, manifest: createManifest()}, Manifest{}, nil
 	}
 	return helpOpenOrCreateManifestFile(opt.Dir, opt.ReadOnly, opt.ExternalMagicVersion,
 		manifestDeletionsRewriteThreshold, opt)
@@ -203,13 +203,15 @@ func (mf *manifestFile) addChanges(changesParam []*pb.ManifestChange, opt Option
 	if err != nil {
 		return err
 	}
-
 	// Maybe we could use O_APPEND instead (on certain file systems)
 	mf.appendLock.Lock()
 	defer mf.appendLock.Unlock()
 	if err := applyChangeSet(&mf.manifest, &changes, opt); err != nil {
 		return err
 	}
+	if mf.inMemory {
+		return nil
+	}
 	// Rewrite manifest if it'd shrink by 1/10 and it's big enough to care
 	if mf.manifest.Deletions > mf.deletionsRewriteThreshold &&
 		mf.manifest.Deletions > manifestDeletionsRatio*(mf.manifest.Creations-mf.manifest.Deletions) {

pb/badgerpb4.pb.go

@@ -69,6 +69,55 @@ func (EncryptionAlgo) EnumDescriptor() ([]byte, []int) {
 	return file_badgerpb4_proto_rawDescGZIP(), []int{0}
 }
 
+type KV_Kind int32
+
+const (
+	KV_KEY      KV_Kind = 0
+	KV_DATA_KEY KV_Kind = 1
+	KV_FILE     KV_Kind = 2
+)
+
+// Enum value maps for KV_Kind.
+var (
+	KV_Kind_name = map[int32]string{
+		0: "KEY",
+		1: "DATA_KEY",
+		2: "FILE",
+	}
+	KV_Kind_value = map[string]int32{
+		"KEY":      0,
+		"DATA_KEY": 1,
+		"FILE":     2,
+	}
+)
+
+func (x KV_Kind) Enum() *KV_Kind {
+	p := new(KV_Kind)
+	*p = x
+	return p
+}
+
+func (x KV_Kind) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (KV_Kind) Descriptor() protoreflect.EnumDescriptor {
+	return file_badgerpb4_proto_enumTypes[1].Descriptor()
+}
+
+func (KV_Kind) Type() protoreflect.EnumType {
+	return &file_badgerpb4_proto_enumTypes[1]
+}
+
+func (x KV_Kind) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use KV_Kind.Descriptor instead.
+func (KV_Kind) EnumDescriptor() ([]byte, []int) {
+	return file_badgerpb4_proto_rawDescGZIP(), []int{0, 0}
+}
+
 type ManifestChange_Operation int32
 
 const (
@@ -99,11 +148,11 @@ func (x ManifestChange_Operation) String() string {
 }
 
 func (ManifestChange_Operation) Descriptor() protoreflect.EnumDescriptor {
-	return file_badgerpb4_proto_enumTypes[1].Descriptor()
+	return file_badgerpb4_proto_enumTypes[2].Descriptor()
 }
 
 func (ManifestChange_Operation) Type() protoreflect.EnumType {
-	return &file_badgerpb4_proto_enumTypes[1]
+	return &file_badgerpb4_proto_enumTypes[2]
 }
 
 func (x ManifestChange_Operation) Number() protoreflect.EnumNumber {
@@ -145,11 +194,11 @@ func (x Checksum_Algorithm) String() string {
 }
 
 func (Checksum_Algorithm) Descriptor() protoreflect.EnumDescriptor {
-	return file_badgerpb4_proto_enumTypes[2].Descriptor()
+	return file_badgerpb4_proto_enumTypes[3].Descriptor()
 }
 
 func (Checksum_Algorithm) Type() protoreflect.EnumType {
-	return &file_badgerpb4_proto_enumTypes[2]
+	return &file_badgerpb4_proto_enumTypes[3]
 }
 
 func (x Checksum_Algorithm) Number() protoreflect.EnumNumber {
@@ -175,7 +224,8 @@ type KV struct {
 	// Stream id is used to identify which stream the KV came from.
 	StreamId uint32 `protobuf:"varint,10,opt,name=stream_id,json=streamId,proto3" json:"stream_id,omitempty"`
 	// Stream done is used to indicate end of stream.
-	StreamDone bool `protobuf:"varint,11,opt,name=stream_done,json=streamDone,proto3" json:"stream_done,omitempty"`
+	StreamDone bool    `protobuf:"varint,11,opt,name=stream_done,json=streamDone,proto3" json:"stream_done,omitempty"`
+	Kind       KV_Kind `protobuf:"varint,12,opt,name=kind,proto3,enum=badgerpb4.KV_Kind" json:"kind,omitempty"`
 }
 
 func (x *KV) Reset() {
@@ -266,6 +316,13 @@ func (x *KV) GetStreamDone() bool {
 	return false
 }
 
+func (x *KV) GetKind() KV_Kind {
+	if x != nil {
+		return x.Kind
+	}
+	return KV_KEY
+}
+
 type KVList struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -642,7 +699,7 @@ var File_badgerpb4_proto protoreflect.FileDescriptor
 
 var file_badgerpb4_proto_rawDesc = []byte{
 	0x0a, 0x0f, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x12, 0x09, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x22, 0xd4, 0x01, 0x0a,
+	0x6f, 0x12, 0x09, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x22, 0xa5, 0x02, 0x0a,
 	0x02, 0x4b, 0x56, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c,
 	0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02,
 	0x20, 0x01, 0x28, 0x0c, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x75,
@@ -656,55 +713,60 @@ var file_badgerpb4_proto_rawDesc = []byte{
 	0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d,
 	0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x5f, 0x64, 0x6f, 0x6e,
 	0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44,
-	0x6f, 0x6e, 0x65, 0x22, 0x44, 0x0a, 0x06, 0x4b, 0x56, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x1d, 0x0a,
-	0x02, 0x6b, 0x76, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x62, 0x61, 0x64, 0x67,
-	0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4b, 0x56, 0x52, 0x02, 0x6b, 0x76, 0x12, 0x1b, 0x0a, 0x09,
-	0x61, 0x6c, 0x6c, 0x6f, 0x63, 0x5f, 0x72, 0x65, 0x66, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52,
-	0x08, 0x61, 0x6c, 0x6c, 0x6f, 0x63, 0x52, 0x65, 0x66, 0x22, 0x48, 0x0a, 0x11, 0x4d, 0x61, 0x6e,
-	0x69, 0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x53, 0x65, 0x74, 0x12, 0x33,
-	0x0a, 0x07, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32,
-	0x19, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4d, 0x61, 0x6e, 0x69,
-	0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x07, 0x63, 0x68, 0x61, 0x6e,
-	0x67, 0x65, 0x73, 0x22, 0x8d, 0x02, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x69, 0x66, 0x65, 0x73, 0x74,
-	0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x49, 0x64, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x04, 0x52, 0x02, 0x49, 0x64, 0x12, 0x33, 0x0a, 0x02, 0x4f, 0x70, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x0e, 0x32, 0x23, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4d,
-	0x61, 0x6e, 0x69, 0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x4f, 0x70,
-	0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x4f, 0x70, 0x12, 0x14, 0x0a, 0x05, 0x4c,
-	0x65, 0x76, 0x65, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05, 0x4c, 0x65, 0x76, 0x65,
-	0x6c, 0x12, 0x15, 0x0a, 0x06, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28,
-	0x04, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x42, 0x0a, 0x0f, 0x65, 0x6e, 0x63, 0x72,
-	0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x61, 0x6c, 0x67, 0x6f, 0x18, 0x05, 0x20, 0x01, 0x28,
-	0x0e, 0x32, 0x19, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x45, 0x6e,
-	0x63, 0x72, 0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x0e, 0x65, 0x6e,
-	0x63, 0x72, 0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x20, 0x0a, 0x0b,
-	0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28,
-	0x0d, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x23,
-	0x0a, 0x09, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0a, 0x0a, 0x06, 0x43,
-	0x52, 0x45, 0x41, 0x54, 0x45, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x45, 0x4c, 0x45, 0x54,
-	0x45, 0x10, 0x01, 0x22, 0x76, 0x0a, 0x08, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x12,
-	0x31, 0x0a, 0x04, 0x61, 0x6c, 0x67, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1d, 0x2e,
-	0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73,
-	0x75, 0x6d, 0x2e, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x52, 0x04, 0x61, 0x6c,
-	0x67, 0x6f, 0x12, 0x10, 0x0a, 0x03, 0x73, 0x75, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52,
-	0x03, 0x73, 0x75, 0x6d, 0x22, 0x25, 0x0a, 0x09, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68,
-	0x6d, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x52, 0x43, 0x33, 0x32, 0x43, 0x10, 0x00, 0x12, 0x0c, 0x0a,
-	0x08, 0x58, 0x58, 0x48, 0x61, 0x73, 0x68, 0x36, 0x34, 0x10, 0x01, 0x22, 0x63, 0x0a, 0x07, 0x44,
-	0x61, 0x74, 0x61, 0x4b, 0x65, 0x79, 0x12, 0x15, 0x0a, 0x06, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x12, 0x0a,
-	0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x64, 0x61, 0x74,
-	0x61, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x76, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02, 0x69,
-	0x76, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74,
-	0x22, 0x42, 0x0a, 0x05, 0x4d, 0x61, 0x74, 0x63, 0x68, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x72, 0x65,
-	0x66, 0x69, 0x78, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x70, 0x72, 0x65, 0x66, 0x69,
-	0x78, 0x12, 0x21, 0x0a, 0x0c, 0x69, 0x67, 0x6e, 0x6f, 0x72, 0x65, 0x5f, 0x62, 0x79, 0x74, 0x65,
-	0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x69, 0x67, 0x6e, 0x6f, 0x72, 0x65, 0x42,
-	0x79, 0x74, 0x65, 0x73, 0x2a, 0x19, 0x0a, 0x0e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x69,
-	0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x07, 0x0a, 0x03, 0x61, 0x65, 0x73, 0x10, 0x00, 0x42,
-	0x23, 0x5a, 0x21, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x64, 0x67,
-	0x72, 0x61, 0x70, 0x68, 0x2d, 0x69, 0x6f, 0x2f, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x2f, 0x76,
-	0x34, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6f, 0x6e, 0x65, 0x12, 0x26, 0x0a, 0x04, 0x6b, 0x69, 0x6e, 0x64, 0x18, 0x0c, 0x20, 0x01, 0x28,
+	0x0e, 0x32, 0x12, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4b, 0x56,
+	0x2e, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x04, 0x6b, 0x69, 0x6e, 0x64, 0x22, 0x27, 0x0a, 0x04, 0x4b,
+	0x69, 0x6e, 0x64, 0x12, 0x07, 0x0a, 0x03, 0x4b, 0x45, 0x59, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08,
+	0x44, 0x41, 0x54, 0x41, 0x5f, 0x4b, 0x45, 0x59, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x46, 0x49,
+	0x4c, 0x45, 0x10, 0x02, 0x22, 0x44, 0x0a, 0x06, 0x4b, 0x56, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x1d,
+	0x0a, 0x02, 0x6b, 0x76, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x62, 0x61, 0x64,
+	0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4b, 0x56, 0x52, 0x02, 0x6b, 0x76, 0x12, 0x1b, 0x0a,
+	0x09, 0x61, 0x6c, 0x6c, 0x6f, 0x63, 0x5f, 0x72, 0x65, 0x66, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04,
+	0x52, 0x08, 0x61, 0x6c, 0x6c, 0x6f, 0x63, 0x52, 0x65, 0x66, 0x22, 0x48, 0x0a, 0x11, 0x4d, 0x61,
+	0x6e, 0x69, 0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x53, 0x65, 0x74, 0x12,
+	0x33, 0x0a, 0x07, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b,
+	0x32, 0x19, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x4d, 0x61, 0x6e,
+	0x69, 0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x07, 0x63, 0x68, 0x61,
+	0x6e, 0x67, 0x65, 0x73, 0x22, 0x8d, 0x02, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x69, 0x66, 0x65, 0x73,
+	0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x49, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x04, 0x52, 0x02, 0x49, 0x64, 0x12, 0x33, 0x0a, 0x02, 0x4f, 0x70, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x0e, 0x32, 0x23, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e,
+	0x4d, 0x61, 0x6e, 0x69, 0x66, 0x65, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x4f,
+	0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x02, 0x4f, 0x70, 0x12, 0x14, 0x0a, 0x05,
+	0x4c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05, 0x4c, 0x65, 0x76,
+	0x65, 0x6c, 0x12, 0x15, 0x0a, 0x06, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x42, 0x0a, 0x0f, 0x65, 0x6e, 0x63,
+	0x72, 0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x61, 0x6c, 0x67, 0x6f, 0x18, 0x05, 0x20, 0x01,
+	0x28, 0x0e, 0x32, 0x19, 0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x45,
+	0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x0e, 0x65,
+	0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x20, 0x0a,
+	0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x06, 0x20, 0x01,
+	0x28, 0x0d, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x72, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x22,
+	0x23, 0x0a, 0x09, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0a, 0x0a, 0x06,
+	0x43, 0x52, 0x45, 0x41, 0x54, 0x45, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x45, 0x4c, 0x45,
+	0x54, 0x45, 0x10, 0x01, 0x22, 0x76, 0x0a, 0x08, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d,
+	0x12, 0x31, 0x0a, 0x04, 0x61, 0x6c, 0x67, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1d,
+	0x2e, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x70, 0x62, 0x34, 0x2e, 0x43, 0x68, 0x65, 0x63, 0x6b,
+	0x73, 0x75, 0x6d, 0x2e, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x52, 0x04, 0x61,
+	0x6c, 0x67, 0x6f, 0x12, 0x10, 0x0a, 0x03, 0x73, 0x75, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04,
+	0x52, 0x03, 0x73, 0x75, 0x6d, 0x22, 0x25, 0x0a, 0x09, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74,
+	0x68, 0x6d, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x52, 0x43, 0x33, 0x32, 0x43, 0x10, 0x00, 0x12, 0x0c,
+	0x0a, 0x08, 0x58, 0x58, 0x48, 0x61, 0x73, 0x68, 0x36, 0x34, 0x10, 0x01, 0x22, 0x63, 0x0a, 0x07,
+	0x44, 0x61, 0x74, 0x61, 0x4b, 0x65, 0x79, 0x12, 0x15, 0x0a, 0x06, 0x6b, 0x65, 0x79, 0x5f, 0x69,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x12,
+	0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x64, 0x61,
+	0x74, 0x61, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x76, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02,
+	0x69, 0x76, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74,
+	0x18, 0x04, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x41,
+	0x74, 0x22, 0x42, 0x0a, 0x05, 0x4d, 0x61, 0x74, 0x63, 0x68, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x72,
+	0x65, 0x66, 0x69, 0x78, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x70, 0x72, 0x65, 0x66,
+	0x69, 0x78, 0x12, 0x21, 0x0a, 0x0c, 0x69, 0x67, 0x6e, 0x6f, 0x72, 0x65, 0x5f, 0x62, 0x79, 0x74,
+	0x65, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x69, 0x67, 0x6e, 0x6f, 0x72, 0x65,
+	0x42, 0x79, 0x74, 0x65, 0x73, 0x2a, 0x19, 0x0a, 0x0e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74,
+	0x69, 0x6f, 0x6e, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x07, 0x0a, 0x03, 0x61, 0x65, 0x73, 0x10, 0x00,
+	0x42, 0x23, 0x5a, 0x21, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x64,
+	0x67, 0x72, 0x61, 0x70, 0x68, 0x2d, 0x69, 0x6f, 0x2f, 0x62, 0x61, 0x64, 0x67, 0x65, 0x72, 0x2f,
+	0x76, 0x34, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -719,31 +781,33 @@ func file_badgerpb4_proto_rawDescGZIP() []byte {
 	return file_badgerpb4_proto_rawDescData
 }
 
-var file_badgerpb4_proto_enumTypes = make([]protoimpl.EnumInfo, 3)
+var file_badgerpb4_proto_enumTypes = make([]protoimpl.EnumInfo, 4)
 var file_badgerpb4_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
 var file_badgerpb4_proto_goTypes = []interface{}{
 	(EncryptionAlgo)(0),           // 0: badgerpb4.EncryptionAlgo
-	(ManifestChange_Operation)(0), // 1: badgerpb4.ManifestChange.Operation
-	(Checksum_Algorithm)(0),       // 2: badgerpb4.Checksum.Algorithm
-	(*KV)(nil),                    // 3: badgerpb4.KV
-	(*KVList)(nil),                // 4: badgerpb4.KVList
-	(*ManifestChangeSet)(nil),     // 5: badgerpb4.ManifestChangeSet
-	(*ManifestChange)(nil),        // 6: badgerpb4.ManifestChange
-	(*Checksum)(nil),              // 7: badgerpb4.Checksum
-	(*DataKey)(nil),               // 8: badgerpb4.DataKey
-	(*Match)(nil),                 // 9: badgerpb4.Match
+	(KV_Kind)(0),                  // 1: badgerpb4.KV.Kind
+	(ManifestChange_Operation)(0), // 2: badgerpb4.ManifestChange.Operation
+	(Checksum_Algorithm)(0),       // 3: badgerpb4.Checksum.Algorithm
+	(*KV)(nil),                    // 4: badgerpb4.KV
+	(*KVList)(nil),                // 5: badgerpb4.KVList
+	(*ManifestChangeSet)(nil),     // 6: badgerpb4.ManifestChangeSet
+	(*ManifestChange)(nil),        // 7: badgerpb4.ManifestChange
+	(*Checksum)(nil),              // 8: badgerpb4.Checksum
+	(*DataKey)(nil),               // 9: badgerpb4.DataKey
+	(*Match)(nil),                 // 10: badgerpb4.Match
 }
 var file_badgerpb4_proto_depIdxs = []int32{
-	3, // 0: badgerpb4.KVList.kv:type_name -> badgerpb4.KV
-	6, // 1: badgerpb4.ManifestChangeSet.changes:type_name -> badgerpb4.ManifestChange
-	1, // 2: badgerpb4.ManifestChange.Op:type_name -> badgerpb4.ManifestChange.Operation
-	0, // 3: badgerpb4.ManifestChange.encryption_algo:type_name -> badgerpb4.EncryptionAlgo
-	2, // 4: badgerpb4.Checksum.algo:type_name -> badgerpb4.Checksum.Algorithm
-	5, // [5:5] is the sub-list for method output_type
-	5, // [5:5] is the sub-list for method input_type
-	5, // [5:5] is the sub-list for extension type_name
-	5, // [5:5] is the sub-list for extension extendee
-	0, // [0:5] is the sub-list for field type_name
+	1, // 0: badgerpb4.KV.kind:type_name -> badgerpb4.KV.Kind
+	4, // 1: badgerpb4.KVList.kv:type_name -> badgerpb4.KV
+	7, // 2: badgerpb4.ManifestChangeSet.changes:type_name -> badgerpb4.ManifestChange
+	2, // 3: badgerpb4.ManifestChange.Op:type_name -> badgerpb4.ManifestChange.Operation
+	0, // 4: badgerpb4.ManifestChange.encryption_algo:type_name -> badgerpb4.EncryptionAlgo
+	3, // 5: badgerpb4.Checksum.algo:type_name -> badgerpb4.Checksum.Algorithm
+	6, // [6:6] is the sub-list for method output_type
+	6, // [6:6] is the sub-list for method input_type
+	6, // [6:6] is the sub-list for extension type_name
+	6, // [6:6] is the sub-list for extension extendee
+	0, // [0:6] is the sub-list for field type_name
 }
 
 func init() { file_badgerpb4_proto_init() }
@@ -842,7 +906,7 @@ func file_badgerpb4_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_badgerpb4_proto_rawDesc,
-			NumEnums:      3,
+			NumEnums:      4,
 			NumMessages:   7,
 			NumExtensions: 0,
 			NumServices:   0,

pb/badgerpb4.proto

@@ -22,6 +22,13 @@ message KV {
   uint32 stream_id = 10;
   // Stream done is used to indicate end of stream.
   bool stream_done = 11;
+
+  enum Kind {
+    KEY = 0;
+    DATA_KEY = 1;
+    FILE = 2;
+  }
+  Kind kind = 12;
 }
 
 message KVList {

stream.go

@@ -8,6 +8,7 @@ package badger
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -17,6 +18,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dgraph-io/badger/v4/pb"
+	"github.com/dgraph-io/badger/v4/table"
 	"github.com/dgraph-io/badger/v4/y"
 	"github.com/dgraph-io/ristretto/v2/z"
 )
@@ -81,7 +83,9 @@ type Stream struct {
 	Send func(buf *z.Buffer) error
 
 	// Read data above the sinceTs. All keys with version =< sinceTs will be ignored.
-	SinceTs      uint64
+	SinceTs uint64
+	// FullCopy should be set to true only when encryption mode is same for sender and receiver.
+	FullCopy     bool
 	readTs       uint64
 	db           *DB
 	rangeCh      chan keyRange
@@ -106,9 +110,6 @@ func (st *Stream) ToList(key []byte, itr *Iterator) (*pb.KVList, error) {
 	list := &pb.KVList{}
 	for ; itr.Valid(); itr.Next() {
 		item := itr.Item()
-		if item.IsDeletedOrExpired() {
-			break
-		}
 		if !bytes.Equal(key, item.Key()) {
 			// Break out on the first encounter with another key.
 			break
@@ -126,6 +127,8 @@ func (st *Stream) ToList(key []byte, itr *Iterator) (*pb.KVList, error) {
 		}
 		kv.Version = item.Version()
 		kv.ExpiresAt = item.ExpiresAt()
+		// As we do full copy, we need to transmit only if it is a delete key or not.
+		kv.Meta = []byte{item.meta & bitDelete}
 		kv.UserMeta = a.Copy([]byte{item.UserMeta()})
 
 		list.Kv = append(list.Kv, kv)
@@ -136,6 +139,12 @@ func (st *Stream) ToList(key []byte, itr *Iterator) (*pb.KVList, error) {
 		if item.DiscardEarlierVersions() {
 			break
 		}
+		if item.IsDeletedOrExpired() {
+			// We do a FullCopy in stream. It might happen that tables from L6 contain K(version=1),
+			// while the table at L4 that was not copied contains K(version=2) with delete mark.
+			// Hence, we need to send the deleted or expired item too.
+			break
+		}
 	}
 	return list, nil
 }
@@ -162,18 +171,10 @@ func (st *Stream) produceRanges(ctx context.Context) {
 }
 
 // produceKVs picks up ranges from rangeCh, generates KV lists and sends them to kvChan.
-func (st *Stream) produceKVs(ctx context.Context, threadId int) error {
+func (st *Stream) produceKVs(ctx context.Context, itr *Iterator) error {
 	st.numProducers.Add(1)
 	defer st.numProducers.Add(-1)
 
-	var txn *Txn
-	if st.readTs > 0 {
-		txn = st.db.NewTransactionAt(st.readTs, false)
-	} else {
-		txn = st.db.NewTransaction(false)
-	}
-	defer txn.Discard()
-
 	// produceKVs is running iterate serially. So, we can define the outList here.
 	outList := z.NewBuffer(2*batchSize, "Stream.ProduceKVs")
 	defer func() {
@@ -183,15 +184,6 @@ func (st *Stream) produceKVs(ctx context.Context, threadId int) error {
 	}()
 
 	iterate := func(kr keyRange) error {
-		iterOpts := DefaultIteratorOptions
-		iterOpts.AllVersions = true
-		iterOpts.Prefix = st.Prefix
-		iterOpts.PrefetchValues = false
-		iterOpts.SinceTs = st.SinceTs
-		itr := txn.NewIterator(iterOpts)
-		itr.ThreadId = threadId
-		defer itr.Close()
-
 		itr.Alloc = z.NewAllocator(1<<20, "Stream.Iterate")
 		defer itr.Alloc.Release()
 
@@ -375,6 +367,77 @@ outer:
 	return nil
 }
 
+func (st *Stream) copyTablesOver(ctx context.Context, tableMatrix [][]*table.Table) error {
+	// TODO: See if making this concurrent would be helpful. Most likely it won't.
+	// But, if it does work, then most like <3 goroutines might be sufficient.
+	infof := st.db.opt.Infof
+	// Make a copy of the manifest so that we don't have race condition.
+	manifest := st.db.manifest.manifest.clone(st.db.opt)
+	dataKeys := make(map[uint64]struct{})
+	// Iterate in reverse order so that the receiver gets the bottommost level first.
+	for i := len(tableMatrix) - 1; i >= 0; i-- {
+		level := i
+		tables := tableMatrix[i]
+		for _, t := range tables {
+			// This table can be picked for copying directly.
+			out := z.NewBuffer(int(t.Size())+1024, "Stream.Table")
+			if dk := t.DataKey(); dk != nil {
+				y.AssertTrue(dk.KeyId != 0)
+				// If we have a legit data key, send it over so the table can be decrypted. The same
+				// data key could have been used to encrypt many tables. Avoid sending it
+				// repeatedly.
+				if _, sent := dataKeys[dk.KeyId]; !sent {
+					infof("Sending data key with ID: %d\n", dk.KeyId)
+					val, err := proto.Marshal(dk)
+					y.Check(err)
+
+					// This would go to key registry in destination.
+					kv := &pb.KV{
+						Value: val,
+						Kind:  pb.KV_DATA_KEY,
+					}
+					KVToBuffer(kv, out)
+					dataKeys[dk.KeyId] = struct{}{}
+				}
+			}
+
+			infof("Sending table ID: %d at level: %d. Size: %s\n",
+				t.ID(), level, humanize.IBytes(uint64(t.Size())))
+			tableManifest := manifest.Tables[t.ID()]
+
+			change := &pb.ManifestChange{
+				Op:    pb.ManifestChange_CREATE,
+				Level: uint32(level),
+				KeyId: tableManifest.KeyID,
+				// Hard coding it, since we're supporting only AES for now.
+				EncryptionAlgo: pb.EncryptionAlgo_aes,
+				Compression:    uint32(tableManifest.Compression),
+			}
+			buf, err := proto.Marshal(change)
+			y.Check(err)
+
+			// We send the table along with level to the destination, so they'd know where to
+			// place the tables. We'd send all the tables first, before we start streaming. So, the
+			// destination DB would write streamed keys one level above.
+			kv := &pb.KV{
+				// Key can be used for MANIFEST.
+				Key:   buf,
+				Value: t.Data,
+				Kind:  pb.KV_FILE,
+			}
+			KVToBuffer(kv, out)
+
+			select {
+			case st.kvChan <- out:
+			case <-ctx.Done():
+				_ = out.Release()
+				return ctx.Err()
+			}
+		}
+	}
+	return nil
+}
+
 // Orchestrate runs Stream. It picks up ranges from the SSTables, then runs NumGo number of
 // goroutines to iterate over these ranges and batch up KVs in lists. It concurrently runs a single
 // goroutine to pick these lists, batch them up further and send to Output.Send. Orchestrate also
@@ -382,6 +445,11 @@ outer:
 // are serial. In case any of these steps encounter an error, Orchestrate would stop execution and
 // return that error. Orchestrate can be called multiple times, but in serial order.
 func (st *Stream) Orchestrate(ctx context.Context) error {
+	if st.FullCopy {
+		if !st.db.opt.managedTxns || st.SinceTs != 0 || st.ChooseKey != nil && st.KeyToList != nil {
+			panic("Got invalid stream options when doing full copy")
+		}
+	}
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	st.rangeCh = make(chan keyRange, 3) // Contains keys for posting lists.
@@ -395,26 +463,6 @@ func (st *Stream) Orchestrate(ctx context.Context) error {
 		st.KeyToList = st.ToList
 	}
 
-	// Picks up ranges from Badger, and sends them to rangeCh.
-	go st.produceRanges(ctx)
-
-	errCh := make(chan error, st.NumGo) // Stores error by consumeKeys.
-	var wg sync.WaitGroup
-	for i := 0; i < st.NumGo; i++ {
-		wg.Add(1)
-
-		go func(threadId int) {
-			defer wg.Done()
-			// Picks up ranges from rangeCh, generates KV lists, and sends them to kvChan.
-			if err := st.produceKVs(ctx, threadId); err != nil {
-				select {
-				case errCh <- err:
-				default:
-				}
-			}
-		}(i)
-	}
-
 	// Pick up key-values from kvChan and send to stream.
 	kvErr := make(chan error, 1)
 	go func() {
@@ -425,6 +473,138 @@ func (st *Stream) Orchestrate(ctx context.Context) error {
 		}
 		kvErr <- err
 	}()
+
+	// Pick all relevant tables from levels. We'd use this to copy them over,
+	// or generate iterators from them.
+	memTables, decr := st.db.getMemTables()
+	defer decr()
+
+	opts := DefaultIteratorOptions
+	opts.Prefix = st.Prefix
+	opts.SinceTs = st.SinceTs
+	tableMatrix := st.db.lc.getTables(&opts)
+	defer func() {
+		for _, tables := range tableMatrix {
+			for _, t := range tables {
+				_ = t.DecrRef()
+			}
+		}
+	}()
+	y.AssertTrue(len(tableMatrix) == st.db.opt.MaxLevels)
+
+	infof := st.db.opt.Infof
+	copyTables := func() error {
+		// Figure out which tables we can copy. Only choose from the last 2 levels.
+		// Say last level has data of size 100. Given a 10x level multiplier and
+		// assuming the tree is balanced, second last level would have 10, and the
+		// third last level would have 1. The third last level would only have 1%
+		// of the data of the last level. It's OK for us to stop there and just
+		// stream it, instead of trying to copy over those tables too.  When we
+		// copy over tables to Level i, we can't stream any data to level i, i+1,
+		// and so on. The stream has to create tables at level i-1, so there can be
+		// overlap between the tables at i-1 and i.
+
+		// Let's pick the tables which can be fully copied over from last level.
+		threshold := len(tableMatrix) - 2
+		toCopy := make([][]*table.Table, len(tableMatrix))
+		var numCopy, numStream int
+		for lev, tables := range tableMatrix {
+			// We stream only the data in the two bottommost levels.
+			if lev < threshold {
+				numStream += len(tables)
+				continue
+			}
+			var rem []*table.Table
+			cp := tables[:0]
+			for _, t := range tables {
+				// We can only copy over those tables that satisfy following conditions:
+				// - All the keys have version less than st.readTs
+				// - st.Prefix fully covers the table
+				if t.MaxVersion() > st.readTs || !t.CoveredByPrefix(st.Prefix) {
+					rem = append(rem, t)
+					continue
+				}
+				cp = append(cp, t)
+			}
+			toCopy[lev] = cp       // Pick tables to copy.
+			tableMatrix[lev] = rem // Keep remaining for streaming.
+			numCopy += len(cp)
+			numStream += len(rem)
+		}
+		infof("Num tables to copy: %d. Num to stream: %d\n", numCopy, numStream)
+
+		return st.copyTablesOver(ctx, toCopy)
+	}
+
+	if st.FullCopy {
+		// As of now, we don't handle the non-zero SinceTs.
+		if err := copyTables(); err != nil {
+			return fmt.Errorf("while copying tables: %w", err)
+		}
+	}
+
+	var txn *Txn
+	if st.readTs > 0 {
+		txn = st.db.NewTransactionAt(st.readTs, false)
+	} else {
+		txn = st.db.NewTransaction(false)
+	}
+	defer txn.Discard()
+
+	newIterator := func(threadId int) *Iterator {
+		var itrs []y.Iterator
+		for _, mt := range memTables {
+			itrs = append(itrs, mt.sl.NewUniIterator(false))
+		}
+		if tables := tableMatrix[0]; len(tables) > 0 {
+			itrs = append(itrs, iteratorsReversed(tables, 0)...)
+		}
+		for _, tables := range tableMatrix[1:] {
+			if len(tables) == 0 {
+				continue
+			}
+			itrs = append(itrs, table.NewConcatIterator(tables, 0))
+		}
+
+		opt := DefaultIteratorOptions
+		opt.AllVersions = true
+		opt.Prefix = st.Prefix
+		opt.PrefetchValues = false
+		opt.SinceTs = st.SinceTs
+
+		res := &Iterator{
+			txn:      txn,
+			iitr:     table.NewMergeIterator(itrs, false),
+			opt:      opt,
+			readTs:   txn.readTs,
+			ThreadId: threadId,
+		}
+		return res
+	}
+
+	// Picks up ranges from Badger, and sends them to rangeCh.
+	// Just for simplicity, we'd consider all the tables for range production.
+	go st.produceRanges(ctx)
+
+	errCh := make(chan error, st.NumGo) // Stores error by consumeKeys.
+	var wg sync.WaitGroup
+	for i := 0; i < st.NumGo; i++ {
+		wg.Add(1)
+
+		go func(threadId int) {
+			defer wg.Done()
+			// Picks up ranges from rangeCh, generates KV lists, and sends them to kvChan.
+			itr := newIterator(threadId)
+			defer itr.Close()
+			if err := st.produceKVs(ctx, itr); err != nil {
+				select {
+				case errCh <- err:
+				default:
+				}
+			}
+		}(i)
+	}
+
 	wg.Wait()        // Wait for produceKVs to be over.
 	close(st.kvChan) // Now we can close kvChan.
 	defer func() {

stream_writer.go

@@ -7,6 +7,7 @@ package badger
 
 import (
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -31,13 +32,18 @@ import (
 // StreamWriter should not be called on in-use DB instances. It is designed only to bootstrap new
 // DBs.
 type StreamWriter struct {
-	writeLock  sync.Mutex
-	db         *DB
-	done       func()
-	throttle   *y.Throttle
-	maxVersion uint64
-	writers    map[uint32]*sortedWriter
-	prevLevel  int
+	writeLock       sync.Mutex
+	db              *DB
+	done            func()
+	throttle        *y.Throttle
+	maxVersion      uint64
+	writers         map[uint32]*sortedWriter
+	prevLevel       int
+	senderPrevLevel int
+	keyId           map[uint64]*pb.DataKey // map stores reader's keyId to data key.
+	// Writer might receive tables first, and then receive keys. If true, that means we have
+	// started processing keys.
+	processingKeys bool
 }
 
 // NewStreamWriter creates a StreamWriter. Right after creating StreamWriter, Prepare must be
@@ -51,6 +57,7 @@ func (db *DB) NewStreamWriter() *StreamWriter {
 		// concurrent streams being processed.
 		throttle: y.NewThrottle(16),
 		writers:  make(map[uint32]*sortedWriter),
+		keyId:    make(map[uint64]*pb.DataKey),
 	}
 }
 
@@ -153,8 +160,64 @@ func (sw *StreamWriter) Write(buf *z.Buffer) error {
 		if _, ok := closedStreams[kv.StreamId]; ok {
 			panic(fmt.Sprintf("write performed on closed stream: %d", kv.StreamId))
 		}
+		switch kv.Kind {
+		case pb.KV_DATA_KEY:
+			sw.writeLock.Lock()
+			defer sw.writeLock.Unlock()
+			y.AssertTrue(len(sw.db.opt.EncryptionKey) > 0)
+			var dk pb.DataKey
+			if err := proto.Unmarshal(kv.Value, &dk); err != nil {
+				return fmt.Errorf("unmarshal failed %s: %w", kv.Value, err)
+			}
+			readerId := dk.KeyId
+			if _, ok := sw.keyId[readerId]; !ok {
+				// Insert the data key to the key registry if not already inserted.
+				id, err := sw.db.registry.AddKey(&dk)
+				if err != nil {
+					return fmt.Errorf("failed to write data key: %w", err)
+				}
+				dk.KeyId = id
+				sw.keyId[readerId] = &dk
+			}
+			return nil
+		case pb.KV_FILE:
+			sw.writeLock.Lock()
+			defer sw.writeLock.Unlock()
+			// All tables should be recieved before any of the keys.
+			if sw.processingKeys {
+				return errors.New("Received pb.KV_FILE after pb.KV_KEY")
+			}
+			var change pb.ManifestChange
+			if err := proto.Unmarshal(kv.Key, &change); err != nil {
+				return fmt.Errorf("unable to unmarshal manifest change: %w", err)
+			}
+			level := int(change.Level)
+			if sw.senderPrevLevel == 0 {
+				// We received the first file, set the sender's and receiver's max levels.
+				sw.senderPrevLevel = level
+				sw.prevLevel = len(sw.db.lc.levels) - 1
+			}
+
+			// This is based on the assumption that the tables from the last
+			// level will be sent first and then the second last level tables.
+			// As long as the kv.Version (which stores the level) is same as
+			// the prevLevel, we know we're processing a last level table.
+			// The last level for this DB can be 8 while the DB that's sending
+			// this could have the last level at 7.
+			if sw.senderPrevLevel != level {
+				// If the previous level and the current level is different, we
+				// must be processing a table from the next last level.
+				sw.senderPrevLevel = level
+				sw.prevLevel--
+			}
+			dk := sw.keyId[change.KeyId]
+			return sw.db.lc.AddTable(&kv, sw.prevLevel, dk, &change)
+		case pb.KV_KEY:
+			// Pass. The following code will handle the keys.
+		}
 
 		sw.writeLock.Lock()
+		sw.processingKeys = true
 		if sw.maxVersion < kv.Version {
 			sw.maxVersion = kv.Version
 		}
@@ -173,6 +236,7 @@ func (sw *StreamWriter) Write(buf *z.Buffer) error {
 		if len(kv.UserMeta) > 0 {
 			userMeta = kv.UserMeta[0]
 		}
+
 		e := &Entry{
 			Key:       y.KeyWithTs(kv.Key, kv.Version),
 			Value:     y.Copy(kv.Value),

table/table.go

@@ -134,10 +134,13 @@ func (t *Table) UncompressedSize() uint32 { return t.cheapIndex().UncompressedSi
 // KeyCount is the total number of keys in this table.
 func (t *Table) KeyCount() uint32 { return t.cheapIndex().KeyCount }
 
-// OnDiskSize returns the total size of key-values stored in this table (including the
-// disk space occupied on the value log).
+// OnDiskSize returns the total size of key-values stored in this table
+// (including the disk space occupied on the value log).
 func (t *Table) OnDiskSize() uint32 { return t.cheapIndex().OnDiskSize }
 
+// DataKey returns the encryption key
+func (t *Table) DataKey() *pb.DataKey { return t.opt.DataKey }
+
 // CompressionType returns the compression algorithm used for block compression.
 func (t *Table) CompressionType() options.CompressionType {
 	return t.opt.Compression
@@ -261,6 +264,33 @@ func CreateTable(fname string, builder *Builder) (*Table, error) {
 	return OpenTable(mf, *builder.opts)
 }
 
+func newFile(fname string, sz int) (*z.MmapFile, error) {
+	mf, err := z.OpenMmapFile(fname, os.O_CREATE|os.O_RDWR|os.O_EXCL, sz)
+	if err == z.NewFile {
+		// Expected.
+	} else if err != nil {
+		return nil, y.Wrapf(err, "while creating table: %s", fname)
+	} else {
+		return nil, errors.Errorf("file already exists: %s", fname)
+	}
+	return mf, nil
+}
+
+func CreateTableFromBuffer(fname string, buf []byte, opts Options) (*Table, error) {
+	mf, err := newFile(fname, len(buf))
+	if err != nil {
+		return nil, err
+	}
+
+	// We cannot use the buf directly here because it is not mmapped.
+	written := copy(mf.Data, buf)
+	y.AssertTrue(written == len(mf.Data))
+	if err := z.Msync(mf.Data); err != nil {
+		return nil, y.Wrapf(err, "while calling msync on %s", fname)
+	}
+	return OpenTable(mf, opts)
+}
+
 // OpenTable assumes file has only one table and opens it. Takes ownership of fd upon function
 // entry. Returns a table with one reference count on it (decrementing which may delete the file!
 // -- consider t.Close() instead). The fd has to writeable because we call Truncate on it before
@@ -681,6 +711,12 @@ func (t *Table) DoesNotHave(hash uint32) bool {
 	return !mayContain
 }
 
+// CoveredByPrefix returns true if all the keys in the table are prefixed by the given prefix.
+func (t *Table) CoveredByPrefix(prefix []byte) bool {
+	return bytes.HasPrefix(y.ParseKey(t.Biggest()), prefix) &&
+		bytes.HasPrefix(y.ParseKey(t.Smallest()), prefix)
+}
+
 // readTableIndex reads table index from the sst and returns its pb format.
 func (t *Table) readTableIndex() (*fb.TableIndex, error) {
 	data := t.readNoFail(t.indexStart, t.indexLen)