reset fragment.rowCache after importValue

Author: travisturner

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.

fragment_internal_test.go

@@ -3363,15 +3363,15 @@ func TestImportMultipleValues(t *testing.T) {
 		cols      []uint64
 		vals      []int64
 		checkCols []uint64
-		checkVals []uint64
+		checkVals []int64
 		depth     uint
 	}{
 		{
 			cols:      []uint64{0, 0},
 			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)

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)