diff --git a/fragment.go b/fragment.go index 03fc11b66..1690e5e38 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 1181d5e82..92b6d107f 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 d61640d76..0b6558511 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)