Remove dictionary rows when no document is referring to it

index/store/boltdb/writer.go

@@ -68,10 +68,23 @@ func (w *Writer) ExecuteBatch(batch store.KVBatch) (err error) {
 			err = fmt.Errorf("merge operator returned failure")
 			return
 		}
-		err = bucket.Put(kb, mergedVal)
+
+		var decodedVal uint64
+		decodedVal, err = w.store.mo.DecodeMergedVal(mergedVal)
 		if err != nil {
 			return
 		}
+
+		if decodedVal > 0 {
+			if err = bucket.Put(kb, mergedVal); err != nil {
+				return
+			}
+		} else {
+			// Delete the shared node when its count reaches 0.
+			if err = bucket.Delete(kb); err != nil {
+				return
+			}
+		}
 	}
 
 	for _, op := range emulatedBatch.Ops {

index/store/goleveldb/writer.go

@@ -55,8 +55,19 @@ func (w *Writer) ExecuteBatch(b store.KVBatch) error {
 		if !fullMergeOk {
 			return fmt.Errorf("merge operator returned failure")
 		}
-		// add the final merge to this batch
-		batch.batch.Put(kb, mergedVal)
+
+		decodedVal, err := w.store.mo.DecodeMergedVal(mergedVal)
+		if err != nil {
+			return err
+		}
+
+		if decodedVal > 0 {
+			// Add the final merge to this batch.
+			batch.batch.Put(kb, mergedVal)
+		} else {
+			// Remove this record when its frequency drops to zero.
+			batch.batch.Delete(kb)
+		}
 	}
 
 	// now execute the batch

index/store/gtreap/writer.go

@@ -55,7 +55,18 @@ func (w *Writer) ExecuteBatch(batch store.KVBatch) error {
 		if !fullMergeOk {
 			return fmt.Errorf("merge operator returned failure")
 		}
-		w.s.t = w.s.t.Upsert(&Item{k: kb, v: mergedVal}, rand.Int())
+
+		decodedVal, err := w.s.mo.DecodeMergedVal(mergedVal)
+		if err != nil {
+			return err
+		}
+
+		if decodedVal > 0 {
+			w.s.t = w.s.t.Upsert(&Item{k: kb, v: mergedVal}, rand.Int())
+		} else {
+			// Delete the shared node when its count reaches 0.
+			w.s.t = w.s.t.Delete(&Item{k: kb})
+		}
 	}
 
 	for _, op := range emulatedBatch.Ops {

index/store/merge.go

@@ -29,6 +29,9 @@ type MergeOperator interface {
 	// all processing until the FullMerge is done.
 	PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool)
 
+	// DecodeMergeVal decodes a merged value.
+	DecodeMergedVal(val []byte) (uint64, error)
+
 	// Name returns an identifier for the operator
 	Name() string
 }

index/store/test/merge.go

@@ -29,16 +29,25 @@ func encodeUint64(in uint64) []byte {
 	return rv
 }
 
+func encodeInt64(in int64) []byte {
+	rv := make([]byte, 8)
+	binary.LittleEndian.PutUint64(rv, uint64(in))
+	return rv
+}
+
 func CommonTestMerge(t *testing.T, s store.KVStore) {
 
-	testKey := []byte("k1")
+	textKey1 := []byte("k1")
+	testKey2 := []byte("k2")
 
 	data := []struct {
 		key []byte
 		val []byte
 	}{
-		{testKey, encodeUint64(1)},
-		{testKey, encodeUint64(1)},
+		{textKey1, encodeUint64(1)},
+		{textKey1, encodeUint64(1)},
+		{testKey2, encodeInt64(1)},
+		{testKey2, encodeInt64(-1)},
 	}
 
 	// open a writer
@@ -70,7 +79,7 @@ func CommonTestMerge(t *testing.T, s store.KVStore) {
 	}
 
 	// read key
-	returnedVal, err := reader.Get(testKey)
+	returnedVal, err := reader.Get(textKey1)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -81,6 +90,16 @@ func CommonTestMerge(t *testing.T, s store.KVStore) {
 		t.Errorf("expected 2, got %d", mergedval)
 	}
 
+	returnedVal, err = reader.Get(testKey2)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// The node is not expected to be inserted when the count is zero.
+	if returnedVal != nil {
+		t.Errorf("expected nil, got %v", binary.LittleEndian.Uint64(returnedVal))
+	}
+
 	// close the reader
 	err = reader.Close()
 	if err != nil {
@@ -117,6 +136,10 @@ func (mc *TestMergeCounter) PartialMerge(key, leftOperand, rightOperand []byte)
 	return rv, true
 }
 
+func (mc *TestMergeCounter) DecodeMergedVal(val []byte) (uint64, error) {
+	return binary.LittleEndian.Uint64(val), nil
+}
+
 func (mc *TestMergeCounter) Name() string {
 	return "test_merge_counter"
 }

index/upsidedown/row_merge.go

@@ -71,6 +71,10 @@ func (m *upsideDownMerge) PartialMerge(key, leftOperand, rightOperand []byte) ([
 	return rv, true
 }
 
+func (m *upsideDownMerge) DecodeMergedVal(val []byte) (uint64, error) {
+	return dictionaryRowParseV(val)
+}
+
 func (m *upsideDownMerge) Name() string {
 	return "upsideDownMerge"
 }