Repository: petermattis/pebble
Pull request #158 (2019-06-04 13:33:25)

Export BatchReader
Export BatchReader so users can have a tool for iterating over the contents
of a batch.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/158)
<!-- Reviewable:end -->

petermattis: @Ryanfsdf can you give this a review? It is a fairly small change, but will introduce you to the internal format of batches.

Patch:
diff --git a/batch.go b/batch.go
index e38eb3a..2a66836 100644
--- a/batch.go
+++ b/batch.go
@@ -258,8 +258,8 @@ func (b *Batch) release() {
 
 func (b *Batch) refreshMemTableSize() {
 	b.memTableSize = 0
-	for iter := b.iter(); ; {
-		_, key, value, ok := iter.next()
+	for r := b.Reader(); ; {
+		_, key, value, ok := r.Next()
 		if !ok {
 			break
 		}
@@ -288,9 +288,9 @@ func (b *Batch) Apply(batch *Batch, _ *WriteOptions) error {
 	count := binary.LittleEndian.Uint32(batch.storage.data[8:12])
 	b.setCount(b.count() + count)
 
-	for iter := batchReader(b.storage.data[offset:]); len(iter) > 0; {
+	for iter := BatchReader(b.storage.data[offset:]); len(iter) > 0; {
 		offset := uintptr(unsafe.Pointer(&iter[0])) - uintptr(unsafe.Pointer(&b.storage.data[0]))
-		kind, key, value, ok := iter.next()
+		kind, key, value, ok := iter.Next()
 		if !ok {
 			break
 		}
@@ -448,8 +448,6 @@ func (b *Batch) DeleteRange(start, end []byte, _ *WriteOptions) error {
 // which makes it useful for testing WAL performance.
 //
 // It is safe to modify the contents of the argument after LogData returns.
-//
-// TODO(peter): untested.
 func (b *Batch) LogData(data []byte, _ *WriteOptions) error {
 	if len(b.storage.data) == 0 {
 		b.init(len(data) + binary.MaxVarintLen64 + batchHeaderLen)
@@ -651,7 +649,9 @@ func (b *Batch) count() uint32 {
 	return binary.LittleEndian.Uint32(b.countData())
 }
 
-func (b *Batch) iter() batchReader {
+// Reader returns a BatchReader for the current batch contents. If the batch is
+// mutated, the new entries will not be visible to the reader.
+func (b *Batch) Reader() BatchReader {
 	return b.storage.data[batchHeaderLen:]
 }
 
@@ -690,11 +690,18 @@ func batchDecodeStr(data []byte) (odata []byte, s []byte, ok bool) {
 	return data[v:], data[:v], true
 }
 
-type batchReader []byte
+// BatchReader iterates over the entries contained in a batch.
+type BatchReader []byte
+
+// MakeBatchReader constructs a BatchReader from a batch representation. The
+// header (containing the batch count and seqnum) is ignored.
+func MakeBatchReader(repr []byte) BatchReader {
+	return repr[batchHeaderLen:]
+}
 
-// next returns the next operation in this batch.
-// The final return value is false if the batch is corrupi.
-func (r *batchReader) next() (kind InternalKeyKind, ukey []byte, value []byte, ok bool) {
+// Next returns the next entry in this batch. The final return value is false
+// if the batch is corrupt. The end of batch is reached when len(r)==0.
+func (r *BatchReader) Next() (kind InternalKeyKind, ukey []byte, value []byte, ok bool) {
 	p := *r
 	if len(p) == 0 {
 		return 0, nil, nil, false
@@ -717,7 +724,7 @@ func (r *batchReader) next() (kind InternalKeyKind, ukey []byte, value []byte, o
 	return kind, ukey, value, true
 }
 
-func (r *batchReader) nextStr() (s []byte, ok bool) {
+func (r *BatchReader) nextStr() (s []byte, ok bool) {
 	p := *r
 	u, numBytes := binary.Uvarint(p)
 	if numBytes <= 0 {
@@ -874,9 +881,9 @@ func newFlushableBatch(batch *Batch, comparer *Comparer) *flushableBatch {
 	}
 
 	var index uint32
-	for iter := batchReader(b.data[batchHeaderLen:]); len(iter) > 0; index++ {
+	for iter := BatchReader(b.data[batchHeaderLen:]); len(iter) > 0; index++ {
 		offset := uintptr(unsafe.Pointer(&iter[0])) - uintptr(unsafe.Pointer(&b.data[0]))
-		kind, key, _, ok := iter.next()
+		kind, key, _, ok := iter.Next()
 		if !ok {
 			break
 		}
diff --git a/mem_table.go b/mem_table.go
index 51f1cb1..0ba7458 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -144,8 +144,8 @@ func (m *memTable) apply(batch *Batch, seqNum uint64) error {
 	var ins arenaskl.Inserter
 	var tombstoneCount uint32
 	startSeqNum := seqNum
-	for iter := batch.iter(); ; seqNum++ {
-		kind, ukey, value, ok := iter.next()
+	for r := batch.Reader(); ; seqNum++ {
+		kind, ukey, value, ok := r.Next()
 		if !ok {
 			break
 		}


Test patch:
diff --git a/batch_test.go b/batch_test.go
index b7243d0..2884740 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -35,21 +35,31 @@ func TestBatch(t *testing.T) {
 		{InternalKeyKindSet, "binarydata", "\x00"},
 		{InternalKeyKindSet, "binarydata", "\xff"},
 		{InternalKeyKindMerge, "merge", "mergedata"},
+		{InternalKeyKindMerge, "merge", ""},
+		{InternalKeyKindMerge, "", ""},
+		{InternalKeyKindRangeDelete, "a", "b"},
+		{InternalKeyKindRangeDelete, "", ""},
+		{InternalKeyKindLogData, "logdata", ""},
+		{InternalKeyKindLogData, "", ""},
 	}
 	var b Batch
 	for _, tc := range testCases {
 		switch tc.kind {
 		case InternalKeyKindSet:
-			b.Set([]byte(tc.key), []byte(tc.value), nil)
+			_ = b.Set([]byte(tc.key), []byte(tc.value), nil)
 		case InternalKeyKindMerge:
-			b.Merge([]byte(tc.key), []byte(tc.value), nil)
+			_ = b.Merge([]byte(tc.key), []byte(tc.value), nil)
 		case InternalKeyKindDelete:
-			b.Delete([]byte(tc.key), nil)
+			_ = b.Delete([]byte(tc.key), nil)
+		case InternalKeyKindRangeDelete:
+			_ = b.DeleteRange([]byte(tc.key), []byte(tc.value), nil)
+		case InternalKeyKindLogData:
+			_ = b.LogData([]byte(tc.key), nil)
 		}
 	}
-	iter := b.iter()
+	r := b.Reader()
 	for _, tc := range testCases {
-		kind, k, v, ok := iter.next()
+		kind, k, v, ok := r.Next()
 		if !ok {
 			t.Fatalf("next returned !ok: test case = %v", tc)
 		}
@@ -59,8 +69,8 @@ func TestBatch(t *testing.T) {
 				kind, key, value, tc.kind, tc.key, tc.value)
 		}
 	}
-	if len(iter) != 0 {
-		t.Errorf("iterator was not exhausted: remaining bytes = %q", iter)
+	if len(r) != 0 {
+		t.Errorf("reader was not exhausted: remaining bytes = %q", r)
 	}
 }