From 7ce7d98cbadeeba22b8ae805a3c0c427a0733d34 Mon Sep 17 00:00:00 2001
From: Steve Yen <steve.yen@gmail.com>
Date: Mon, 11 Jan 2016 16:52:07 -0800
Subject: [PATCH] upside_down merge dictionary deltas before using
 batch.Merge()

This change performs more dictionary delta incr/decr math in
batchRows() instead of in the KVStore ExecuteBatch() machinery.
---
 index/upside_down/upside_down.go | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/index/upside_down/upside_down.go b/index/upside_down/upside_down.go
index e195d676..5b66f2be 100644
--- a/index/upside_down/upside_down.go
+++ b/index/upside_down/upside_down.go
@@ -10,6 +10,7 @@
 package upside_down
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -152,6 +153,8 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 	// buffer to work with
 	rowBuf := GetRowBuffer()
 
+	dictionaryDeltas := make(map[string]int64)
+
 	// add
 	for _, row := range addRows {
 		tfr, ok := row.(*TermFrequencyRow)
@@ -163,7 +166,7 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 			if err != nil {
 				return err
 			}
-			wb.Merge(rowBuf[:dictKeySize], dictionaryTermIncr)
+			dictionaryDeltas[string(rowBuf[:dictKeySize])] += 1
 		}
 		if row.KeySize()+row.ValueSize() > len(rowBuf) {
 			rowBuf = make([]byte, row.KeySize()+row.ValueSize())
@@ -204,7 +207,7 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 			if err != nil {
 				return err
 			}
-			wb.Merge(rowBuf[:dictKeySize], dictionaryTermDecr)
+			dictionaryDeltas[string(rowBuf[:dictKeySize])] -= 1
 		}
 		if row.KeySize()+row.ValueSize() > len(rowBuf) {
 			rowBuf = make([]byte, row.KeySize()+row.ValueSize())
@@ -216,6 +219,15 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 		wb.Delete(rowBuf[:keySize])
 	}
 
+	if 8 > len(rowBuf) {
+		rowBuf = make([]byte, 8)
+	}
+
+	for dictRowKey, delta := range dictionaryDeltas {
+		binary.LittleEndian.PutUint64(rowBuf, uint64(delta))
+		wb.Merge([]byte(dictRowKey), rowBuf[0:8])
+	}
+
 	PutRowBuffer(rowBuf)
 
 	// write out the batch