scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix off-by-one maxRowID in block limits

Browse source 
In the case where a block merge needed to occur
on a replica containing a row on the edge of the block,
the existing logic would inadvertently clear the first
row in the next block. This PR fixes that.
This commit is contained in:
Travis 2020-05-01 15:12:41 -05:00
parent 8e0a787ec0
commit b25796f532
2 changed files with 59 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
fragment.go
View file

@ -1776,8 +1776,8 @@ func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, e
clears = make([]pairSet, len(data)+1)
// Limit upper row/column pair.
maxRowID := uint64(id+1) * HashBlockSize
maxColumnID := uint64(ShardWidth)
maxRowID := (uint64(id+1) * HashBlockSize) - 1
maxColumnID := uint64(ShardWidth) - 1
// Create buffered iterator for local block.
itrs := make([]*bufIterator, 1, len(data)+1)
@ -1857,8 +1857,8 @@ func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, e
sets[i].rowIDs = append(sets[i].rowIDs, min.rowID)
sets[i].columnIDs = append(sets[i].columnIDs, min.columnID)
} else {
clears[i].rowIDs = append(sets[i].rowIDs, min.rowID)
clears[i].columnIDs = append(sets[i].columnIDs, min.columnID)
clears[i].rowIDs = append(clears[i].rowIDs, min.rowID)
clears[i].columnIDs = append(clears[i].columnIDs, min.columnID)
}
}
}

55
holder_test.go
View file

@ -490,6 +490,61 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
}
}
// Ensure holder can sync with a remote holder and respects
// the row boundaries of the block.
func TestHolderSyncer_BlockIteratorLimits(t *testing.T) {
c := test.MustNewCluster(t, 3)
c[0].Config.Cluster.ReplicaN = 3
c[0].Config.AntiEntropy.Interval = 0
c[1].Config.Cluster.ReplicaN = 3
c[1].Config.AntiEntropy.Interval = 0
err := c.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
defer c.Close()
_, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index i: %v", err)
}
_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize))
if err != nil {
t.Fatalf("creating field f: %v", err)
}
hldr0 := &test.Holder{Holder: c[0].Server.Holder()}
hldr1 := &test.Holder{Holder: c[1].Server.Holder()}
hldr2 := &test.Holder{Holder: c[2].Server.Holder()}
// Set data on the local holder.
hldr0.SetBit("i", "f", 99, 10)
hldr0.SetBit("i", "f", 100, 20)
// Set the same data on one of the replicas
// so that we have a quorum.
hldr1.SetBit("i", "f", 99, 10)
hldr1.SetBit("i", "f", 100, 20)
// Leave the third replica empty to force a block merge.
//
err = c[0].Server.SyncData()
if err != nil {
t.Fatalf("syncing node 0: %v", err)
}
// Verify data is the same on both nodes.
for i, hldr := range []*test.Holder{hldr0, hldr1, hldr2} {
if a := hldr.Row("i", "f", 99).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
t.Errorf("unexpected columns(%d/0): %+v", i, a)
}
if a := hldr.Row("i", "f", 100).Columns(); !reflect.DeepEqual(a, []uint64{20}) {
t.Errorf("unexpected columns(%d/0): %+v", i, a)
}
}
}
// Ensure holder can sync time quantum views with a remote holder.
func TestHolderSyncer_TimeQuantum(t *testing.T) {
c := test.MustNewCluster(t, 2)