From 5d71f04e54a337c86d7c5a7173830b0f2e3befc0 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Tue, 5 Nov 2019 17:56:59 -0600 Subject: [PATCH 1/2] reset fragment.rowCache after importValue --- fragment.go | 12 ++++++- fragment_internal_test.go | 66 +++++++++++++++++++++++++++++++++++++-- http/client_test.go | 66 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 140 insertions(+), 4 deletions(-) diff --git a/fragment.go b/fragment.go index ad43b2503..cae9a5caa 100644 --- a/fragment.go +++ b/fragment.go @@ -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. diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 333f15b42..04682da03 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -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) diff --git a/http/client_test.go b/http/client_test.go index b4c32932d..602363078 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -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) From 5ebd7e8da8b10bf91f380d5bdbae669212488e3a Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 11 Nov 2019 16:48:14 -0600 Subject: [PATCH 2/2] move requiredDepth calculation after min/max ranges are checked Without this, a data set with a ludicrously large value in it could break a BSI field's depth even though the import would then reject it. always treat BSI fields as having at least their depth: If you imported only small values, BSI fields could end up not bothering to clear higher bits in existing values, which produced strange behaviors. --- field.go | 58 ++++++++++++++++++++++++++++++-------------------------- 1 file changed, 31 insertions(+), 27 deletions(-) diff --git a/field.go b/field.go index 39c6c5363..a4b36e8d4 100644 --- a/field.go +++ b/field.go @@ -1269,34 +1269,12 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO return errors.Wrap(ErrBSIGroupNotFound, f.name) } - // Find the lowest/highest values. + // We want to determine the required bit depth, in case the field doesn't + // have as many bits currently as would be needed to represent these values, + // but only if the values are in-range for the field. var min, max int64 - for i, value := range values { - if i == 0 || value < min { - min = value - } - if i == 0 || value > max { - max = value - } - } - - // Determine the highest bit depth required by the min & max. - requiredDepth := bitDepthInt64(min - bsig.Base) - if v := bitDepthInt64(max - bsig.Base); v > requiredDepth { - requiredDepth = v - } - - // Increase bit depth if required. - if requiredDepth > bsig.BitDepth { - if err := func() error { - f.mu.Lock() - defer f.mu.Unlock() - bsig.BitDepth = requiredDepth - f.options.BitDepth = requiredDepth - return f.saveMeta() - }(); err != nil { - return errors.Wrap(err, "increasing bsi bit depth") - } + if len(values) > 0 { + min, max = values[0], values[0] } // Split import data by fragment. @@ -1308,6 +1286,12 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO } else if value < bsig.Min { return fmt.Errorf("%v, columnID=%v, value=%v", ErrBSIGroupValueTooLow, columnID, value) } + if value > max { + max = value + } + if value < min { + min = value + } // Attach value to each bsiGroup view. for _, name := range []string{viewName} { @@ -1319,6 +1303,26 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO } } + // Determine the highest bit depth required by the min & max. + requiredDepth := bitDepthInt64(min - bsig.Base) + if v := bitDepthInt64(max - bsig.Base); v > requiredDepth { + requiredDepth = v + } + // Increase bit depth if required. + if requiredDepth > bsig.BitDepth { + if err := func() error { + f.mu.Lock() + defer f.mu.Unlock() + bsig.BitDepth = requiredDepth + f.options.BitDepth = requiredDepth + return f.saveMeta() + }(); err != nil { + return errors.Wrap(err, "increasing bsi bit depth") + } + } else { + requiredDepth = bsig.BitDepth + } + // Import into each fragment. for key, data := range dataByFragment { // The view must already exist (i.e. we can't create it)