From 1a44f02e3caa01261aab844e9747aba96ae95098 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 d6518ea60..fa66847b4 100644 --- a/fragment.go +++ b/fragment.go @@ -2206,7 +2206,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. @@ -2245,6 +2252,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 215a3bd7c..1582ecd3c 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3378,7 +3378,7 @@ func TestImportMultipleValues(t *testing.T) { cols []uint64 vals []int64 checkCols []uint64 - checkVals []uint64 + checkVals []int64 depth uint }{ { @@ -3386,7 +3386,7 @@ func TestImportMultipleValues(t *testing.T) { vals: []int64{97, 100}, depth: 7, checkCols: []uint64{0}, - checkVals: []uint64{100}, + checkVals: []int64{100}, }, } @@ -3410,7 +3410,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) } } @@ -3420,6 +3420,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 e7130c68f..a3a0babbd 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 a804a0dfb17c92ba1ce481ad4ce5494ac7a498f1 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 11 Nov 2019 12:40:14 -0600 Subject: [PATCH 2/2] 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. We also move the computation of requiredDepth, and the change to the field, down, combining it with the other checks of the values for min/max being in range. 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. --- field.go | 58 ++++++++++++++++++++++++++++++-------------------------- 1 file changed, 31 insertions(+), 27 deletions(-) diff --git a/field.go b/field.go index f38668c70..dfc098237 100644 --- a/field.go +++ b/field.go @@ -1409,34 +1409,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. @@ -1448,6 +1426,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} { @@ -1459,6 +1443,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)