From b25796f532c973edb64867cbe01c9fc6522d31fd Mon Sep 17 00:00:00 2001 From: Travis Date: Fri, 1 May 2020 15:12:41 -0500 Subject: [PATCH 1/3] Fix off-by-one maxRowID in block limits 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. --- fragment.go | 8 ++++---- holder_test.go | 55 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/fragment.go b/fragment.go index 78ef0597f..6a801bc50 100644 --- a/fragment.go +++ b/fragment.go @@ -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) } } } diff --git a/holder_test.go b/holder_test.go index d9d11f3f6..3a78080a4 100644 --- a/holder_test.go +++ b/holder_test.go @@ -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) From 1f308066a75634b8ff0d864e8bf50e88d7d27c91 Mon Sep 17 00:00:00 2001 From: Travis Date: Fri, 1 May 2020 15:54:56 -0500 Subject: [PATCH 2/3] fix test error messages --- holder_test.go | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/holder_test.go b/holder_test.go index 3a78080a4..11b537967 100644 --- a/holder_test.go +++ b/holder_test.go @@ -513,18 +513,20 @@ func TestHolderSyncer_BlockIteratorLimits(t *testing.T) { t.Fatalf("creating field f: %v", err) } + blockEdge := uint64(pilosa.HashBlockSize) + 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) + hldr0.SetBit("i", "f", blockEdge-1, 10) + hldr0.SetBit("i", "f", blockEdge, 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) + hldr1.SetBit("i", "f", blockEdge-1, 10) + hldr1.SetBit("i", "f", blockEdge, 20) // Leave the third replica empty to force a block merge. // @@ -536,11 +538,11 @@ func TestHolderSyncer_BlockIteratorLimits(t *testing.T) { // 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", blockEdge-1).Columns(); !reflect.DeepEqual(a, []uint64{10}) { + t.Errorf("unexpected columns(%d/block 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) + if a := hldr.Row("i", "f", blockEdge).Columns(); !reflect.DeepEqual(a, []uint64{20}) { + t.Errorf("unexpected columns(%d/block 1): %+v", i, a) } } } From de0785d30526ff3eafcc2cd4d6851844a93c6566 Mon Sep 17 00:00:00 2001 From: Travis Date: Fri, 1 May 2020 16:15:42 -0500 Subject: [PATCH 3/3] add a test for the "clears" bug --- holder_test.go | 51 +++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/holder_test.go b/holder_test.go index 11b537967..42c010079 100644 --- a/holder_test.go +++ b/holder_test.go @@ -536,7 +536,7 @@ func TestHolderSyncer_BlockIteratorLimits(t *testing.T) { t.Fatalf("syncing node 0: %v", err) } - // Verify data is the same on both nodes. + // Verify data is the same on all nodes. for i, hldr := range []*test.Holder{hldr0, hldr1, hldr2} { if a := hldr.Row("i", "f", blockEdge-1).Columns(); !reflect.DeepEqual(a, []uint64{10}) { t.Errorf("unexpected columns(%d/block 0): %+v", i, a) @@ -547,6 +547,55 @@ func TestHolderSyncer_BlockIteratorLimits(t *testing.T) { } } +// Ensure holder correctly handles clears during block sync. +func TestHolderSyncer_Clears(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 that should be cleared + // because it's the only instance of this value. + hldr0.SetBit("i", "f", 0, 30) + + // Set similar data on the replicas, but + // different from what's on local. This should end + // up being set on all replicas + hldr1.SetBit("i", "f", 0, 20) + hldr2.SetBit("i", "f", 0, 20) + + err = c[0].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 0: %v", err) + } + + // Verify data is the same on all nodes. + for i, hldr := range []*test.Holder{hldr0, hldr1, hldr2} { + if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{20}) { + t.Errorf("unexpected columns(%d): %+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)