From faceecf87b5bd8f8af06c45a77bd21594ef00dd6 Mon Sep 17 00:00:00 2001
From: Marty Schoch <marty.schoch@gmail.com>
Date: Mon, 19 Oct 2015 12:03:38 -0400
Subject: [PATCH] make row buffer size constant/configurable also handle case
 where it is insufficiently sized

---
 index/upside_down/upside_down.go | 24 +++++++++++++++++++++++-
 1 file changed, 23 insertions(+), 1 deletion(-)

diff --git a/index/upside_down/upside_down.go b/index/upside_down/upside_down.go
index c4d10b40..bbe7b443 100644
--- a/index/upside_down/upside_down.go
+++ b/index/upside_down/upside_down.go
@@ -28,6 +28,13 @@ import (
 
 const Name = "upside_down"
 
+// RowBufferSize should ideally this is sized to be the smallest
+// size that can cotain an index row key and its corresponding
+// value.  It is not a limit, if need be a larger buffer is
+// allocated, but performance will be more optimal if *most*
+// rows fit this size.
+const RowBufferSize = 4 * 1024
+
 var VersionKey = []byte{'v'}
 
 var UnsafeBatchUseDetected = fmt.Errorf("bleve.Batch is NOT thread-safe, modification after execution detected")
@@ -120,7 +127,7 @@ func GetRowBuffer() []byte {
 	if rb, ok := rowBufferPool.Get().([]byte); ok {
 		return rb
 	} else {
-		return make([]byte, 4*1024)
+		return make([]byte, RowBufferSize)
 	}
 }
 
@@ -140,12 +147,18 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 	for _, row := range addRows {
 		tfr, ok := row.(*TermFrequencyRow)
 		if ok {
+			if tfr.DictionaryRowKeySize() > len(rowBuf) {
+				rowBuf = make([]byte, tfr.DictionaryRowKeySize())
+			}
 			dictKeySize, err := tfr.DictionaryRowKeyTo(rowBuf)
 			if err != nil {
 				return err
 			}
 			wb.Merge(rowBuf[:dictKeySize], dictionaryTermIncr)
 		}
+		if row.KeySize()+row.ValueSize() > len(rowBuf) {
+			rowBuf = make([]byte, row.KeySize()+row.ValueSize())
+		}
 		keySize, err := row.KeyTo(rowBuf)
 		if err != nil {
 			return err
@@ -156,6 +169,9 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 
 	// update
 	for _, row := range updateRows {
+		if row.KeySize()+row.ValueSize() > len(rowBuf) {
+			rowBuf = make([]byte, row.KeySize()+row.ValueSize())
+		}
 		keySize, err := row.KeyTo(rowBuf)
 		if err != nil {
 			return err
@@ -172,12 +188,18 @@ func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDow
 		tfr, ok := row.(*TermFrequencyRow)
 		if ok {
 			// need to decrement counter
+			if tfr.DictionaryRowKeySize() > len(rowBuf) {
+				rowBuf = make([]byte, tfr.DictionaryRowKeySize())
+			}
 			dictKeySize, err := tfr.DictionaryRowKeyTo(rowBuf)
 			if err != nil {
 				return err
 			}
 			wb.Merge(rowBuf[:dictKeySize], dictionaryTermDecr)
 		}
+		if row.KeySize()+row.ValueSize() > len(rowBuf) {
+			rowBuf = make([]byte, row.KeySize()+row.ValueSize())
+		}
 		keySize, err := row.KeyTo(rowBuf)
 		if err != nil {
 			return err