Skip to content

Commit

Permalink
release-3.5: backport #17263 Fix tx buffer inconsistency if there are…
Browse files Browse the repository at this point in the history
… unordered key writes in one tx.

#17263

Notes:
1. batch_tx_test.go is not incorporated into this backport because main and release-3.5 diverges a lot on buckets/schema/etc..
It is also not the core of the bug fix.
2. verify in tx_buffer.go is removed as mainly due to the dependency is not there in relese-3.5.

Signed-off-by: Chao Chen <[email protected]>
  • Loading branch information
chaochn47 committed Nov 11, 2024
1 parent 4726460 commit c7b58bc
Show file tree
Hide file tree
Showing 2 changed files with 84 additions and 4 deletions.
17 changes: 13 additions & 4 deletions server/mvcc/backend/tx_buffer.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ func (txw *txWriteBuffer) put(bucket Bucket, k, v []byte) {
}

func (txw *txWriteBuffer) putSeq(bucket Bucket, k, v []byte) {
// TODO: Add (in tests?) verification whether k>b[len(b)]
// putSeq is only be called for the data in the Key bucket. The keys
// in the Key bucket should be monotonically increasing revisions.
txw.putInternal(bucket, k, v)
}

Expand Down Expand Up @@ -80,6 +81,9 @@ func (txw *txWriteBuffer) writeback(txr *txReadBuffer) {
rb, ok := txr.buckets[k]
if !ok {
delete(txw.buckets, k)
if seq, ok := txw.bucket2seq[k]; ok && !seq {
wb.dedupe()
}
txr.buckets[k] = wb
continue
}
Expand Down Expand Up @@ -148,7 +152,7 @@ func newBucketBuffer() *bucketBuffer {
func (bb *bucketBuffer) Range(key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
f := func(i int) bool { return bytes.Compare(bb.buf[i].key, key) >= 0 }
idx := sort.Search(bb.used, f)
if idx < 0 {
if idx < 0 || idx >= bb.used {
return nil, nil
}
if len(endKey) == 0 {
Expand Down Expand Up @@ -201,10 +205,15 @@ func (bb *bucketBuffer) merge(bbsrc *bucketBuffer) {
if bytes.Compare(bb.buf[(bb.used-bbsrc.used)-1].key, bbsrc.buf[0].key) < 0 {
return
}
bb.dedupe()
}

// dedupe removes duplicates, using only newest update
func (bb *bucketBuffer) dedupe() {
if bb.used <= 1 {
return
}
sort.Stable(bb)

// remove duplicates, using only newest update
widx := 0
for ridx := 1; ridx < bb.used; ridx++ {
if !bytes.Equal(bb.buf[ridx].key, bb.buf[widx].key) {
Expand Down
71 changes: 71 additions & 0 deletions server/mvcc/backend/tx_buffer_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
// Copyright 2023 The etcd Authors
//
// 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 backend

import (
"testing"

"github.com/stretchr/testify/assert"
)

func TestDedupe(t *testing.T) {
tests := []struct {
name string
keys, vals, expectedKeys, expectedVals []string
}{
{
name: "empty",
keys: []string{},
vals: []string{},
expectedKeys: []string{},
expectedVals: []string{},
},
{
name: "single kv",
keys: []string{"key1"},
vals: []string{"val1"},
expectedKeys: []string{"key1"},
expectedVals: []string{"val1"},
},
{
name: "duplicate key",
keys: []string{"key1", "key1"},
vals: []string{"val1", "val2"},
expectedKeys: []string{"key1"},
expectedVals: []string{"val2"},
},
{
name: "unordered keys",
keys: []string{"key3", "key1", "key4", "key2", "key1", "key4"},
vals: []string{"val1", "val5", "val3", "val4", "val2", "val6"},
expectedKeys: []string{"key1", "key2", "key3", "key4"},
expectedVals: []string{"val2", "val4", "val1", "val6"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
bb := &bucketBuffer{buf: make([]kv, 10), used: 0}
for i := 0; i < len(tt.keys); i++ {
bb.add([]byte(tt.keys[i]), []byte(tt.vals[i]))
}
bb.dedupe()
assert.Len(t, tt.expectedKeys, bb.used)
for i := 0; i < bb.used; i++ {
assert.Equal(t, bb.buf[i].key, []byte(tt.expectedKeys[i]))
assert.Equal(t, bb.buf[i].val, []byte(tt.expectedVals[i]))
}
})
}
}

0 comments on commit c7b58bc

Please sign in to comment.