0
0

fix row merging bug

trying to be clever, we reused the memory allocated for the left
operand when doing partial merges
this had been tested to be safe, in general.  however, the
implementation was then written such that we always reused
globally defined operands, this meant that we mutated
the operands which were intended to always represent
+1/-1
this then cascades quickly to making increment/decrement
values much larger/smaller than they should be
related to #197
This commit is contained in:
Marty Schoch 2015-05-06 11:00:04 -04:00
parent 30a0ba1f9b
commit 57358088ec
2 changed files with 55 additions and 2 deletions

View File

@ -61,8 +61,9 @@ func (m *upsideDownMerge) FullMerge(key, existingValue []byte, operands [][]byte
func (m *upsideDownMerge) PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool) {
left := int64(binary.LittleEndian.Uint64(leftOperand))
right := int64(binary.LittleEndian.Uint64(rightOperand))
binary.LittleEndian.PutUint64(leftOperand, uint64(left+right))
return leftOperand, true
rv := make([]byte, 8)
binary.LittleEndian.PutUint64(rv, uint64(left+right))
return rv, true
}
func (m *upsideDownMerge) Name() string {

View File

@ -0,0 +1,52 @@
// Copyright (c) 2014 Couchbase, Inc.
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
// except in compliance with the License. You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software distributed under the
// License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
// either express or implied. See the License for the specific language governing permissions
// and limitations under the License.
package upside_down
import (
"bytes"
"encoding/binary"
"testing"
)
func TestPartialMerge(t *testing.T) {
tests := []struct {
in [][]byte
out uint64
}{
{
in: [][]byte{dictionaryTermIncr, dictionaryTermIncr, dictionaryTermIncr, dictionaryTermIncr, dictionaryTermIncr},
out: 5,
},
}
mo := &upsideDownMerge{}
for _, test := range tests {
curr := test.in[0]
for _, next := range test.in[1:] {
var ok bool
curr, ok = mo.PartialMerge([]byte("key"), curr, next)
if !ok {
t.Errorf("expected partial merge ok")
}
}
actual := decodeCount(curr)
if actual != test.out {
t.Errorf("expected %d, got %d", test.out, actual)
}
}
}
func decodeCount(in []byte) uint64 {
buf := bytes.NewBuffer(in)
count, _ := binary.ReadUvarint(buf)
return count
}