reset fragment.rowCache after importValue

This commit is contained in:
Travis Turner 2019-11-05 17:56:59 -06:00
parent 9d7ef20206
commit 020b72abbf
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
3 changed files with 140 additions and 4 deletions

View file

@ -2191,7 +2191,14 @@ func (f *fragment) importValueSmallWrite(columnIDs []uint64, values []int64, bit
rowSet[uint64(i)] = struct{}{}
}
err := f.importPositions(toSet, toClear, rowSet)
return errors.Wrap(err, "importing positions")
if err != nil {
return errors.Wrap(err, "importing positions")
}
// Reset the rowCache.
f.rowCache = &simpleCache{make(map[uint64]*Row)}
return nil
}
// importValue bulk imports a set of range-encoded values.
@ -2230,6 +2237,9 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
// We don't actually care, except we want our stats to be accurate.
f.incrementOpN(totalChanges)
// Reset the rowCache.
f.rowCache = &simpleCache{make(map[uint64]*Row)}
// in theory, this should probably have happened anyway, but if enough
// of the bits matched existing bits, we'll be under our opN estimate, and
// we want to ensure that the snapshot happens.

View file

@ -3363,7 +3363,7 @@ func TestImportMultipleValues(t *testing.T) {
cols []uint64
vals []int64
checkCols []uint64
checkVals []uint64
checkVals []int64
depth uint
}{
{
@ -3371,7 +3371,7 @@ func TestImportMultipleValues(t *testing.T) {
vals: []int64{97, 100},
depth: 7,
checkCols: []uint64{0},
checkVals: []uint64{100},
checkVals: []int64{100},
},
}
@ -3395,7 +3395,7 @@ func TestImportMultipleValues(t *testing.T) {
if !exists {
t.Errorf("column %d should exist", cc)
}
if n != 100 {
if n != cv {
t.Errorf("wrong value: %d is not %d", n, cv)
}
}
@ -3405,6 +3405,66 @@ func TestImportMultipleValues(t *testing.T) {
}
}
func TestImportValueRowCache(t *testing.T) {
type testCase struct {
cols []uint64
vals []int64
checkCols []uint64
depth uint
}
tests := []struct {
tc1 testCase
tc2 testCase
}{
{
tc1: testCase{
cols: []uint64{2},
vals: []int64{1},
depth: 1,
checkCols: []uint64{2},
},
tc2: testCase{
cols: []uint64{1000},
vals: []int64{1},
depth: 1,
checkCols: []uint64{2, 1000},
},
},
}
for i, test := range tests {
for _, maxOpN := range []int{1, 10000} {
t.Run(fmt.Sprintf("%dMaxOpN%d", i, maxOpN), func(t *testing.T) {
f := mustOpenBSIFragment("i", "f", viewBSIGroupPrefix+"foo", 0)
f.MaxOpN = maxOpN
defer f.Clean(t)
// First import (tc1)
if err := f.importValue(test.tc1.cols, test.tc1.vals, test.tc1.depth, false); err != nil {
t.Fatalf("importing values: %v", err)
}
if r, err := f.rangeOp(pql.GT, test.tc1.depth, 0); err != nil {
t.Error("getting range of values")
} else if !reflect.DeepEqual(r.Columns(), test.tc1.checkCols) {
t.Errorf("wrong column values. expected: %v, but got: %v", test.tc1.checkCols, r.Columns())
}
// Second import (tc2)
if err := f.importValue(test.tc2.cols, test.tc2.vals, test.tc2.depth, false); err != nil {
t.Fatalf("importing values: %v", err)
}
if r, err := f.rangeOp(pql.GT, test.tc2.depth, 0); err != nil {
t.Error("getting range of values")
} else if !reflect.DeepEqual(r.Columns(), test.tc2.checkCols) {
t.Errorf("wrong column values. expected: %v, but got: %v", test.tc2.checkCols, r.Columns())
}
})
}
}
}
func TestFragmentConcurrentReadWrite(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, CacheTypeRanked)
defer f.Clean(t)

View file

@ -777,6 +777,72 @@ func TestClient_ImportKeys(t *testing.T) {
})
}
func TestClient_ImportIDs(t *testing.T) {
// Ensure that running a query between two imports does
// not affect the result set. It turns out, this is caused
// by the fragment.rowCache failing to be cleared after an
// importValue. This ensures that the rowCache is cleared
// after an import.
t.Run("ImportRangeImport", func(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
idxName := "i"
fldName := "f"
// Load bitmap into cache to ensure cache gets updated.
index := hldr.MustCreateIndexIfNotExists(idxName, pilosa.IndexOptions{Keys: false})
_, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-10000, 10000))
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 2, Value: 1},
}); err != nil {
t.Fatal(err)
}
// Verify range.
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`Row(%s>0)`, fldName),
Remote: false,
}
if result, err := c.Query(context.Background(), idxName, queryRequest); err != nil {
t.Fatal(err)
} else {
res := result.Results[0].(*pilosa.Row).Columns()
if !reflect.DeepEqual(res, []uint64{2}) {
t.Fatalf("unexpected column ids: %v", res)
}
}
// Send import request.
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 1000, Value: 1},
}); err != nil {
t.Fatal(err)
}
// Verify range.
if result, err := c.Query(context.Background(), idxName, queryRequest); err != nil {
t.Fatal(err)
} else {
res := result.Results[0].(*pilosa.Row).Columns()
if !reflect.DeepEqual(res, []uint64{2, 1000}) {
t.Fatalf("unexpected column ids: %v", res)
}
}
})
}
// Ensure client can bulk import value data.
func TestClient_ImportValue(t *testing.T) {
cluster := test.MustRunCluster(t, 1)