Merge pull request #2127 from travisturner/backport-pr2084

Backport pr2084

Author: travisturner

field.go

@@ -1269,23 +1269,45 @@ 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 len(values) > 0 {
+		min, max = values[0], values[0]
+	}
+
+	// Split import data by fragment.
+	dataByFragment := make(map[importKey]importValueData)
+	for i := range columnIDs {
+		columnID, value := columnIDs[i], values[i]
+		if value > bsig.Max {
+			return fmt.Errorf("%v, columnID=%v, value=%v", ErrBSIGroupValueTooHigh, columnID, value)
+		} else if value < bsig.Min {
+			return fmt.Errorf("%v, columnID=%v, value=%v", ErrBSIGroupValueTooLow, columnID, value)
 		}
-		if i == 0 || value > max {
+		if value > max {
 			max = value
 		}
+		if value < min {
+			min = value
+		}
+
+		// Attach value to each bsiGroup view.
+		for _, name := range []string{viewName} {
+			key := importKey{View: name, Shard: columnID / ShardWidth}
+			data := dataByFragment[key]
+			data.ColumnIDs = append(data.ColumnIDs, columnID)
+			data.Values = append(data.Values, value)
+			dataByFragment[key] = data
+		}
 	}
 
 	// 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 {
@@ -1297,26 +1319,8 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
 		}(); err != nil {
 			return errors.Wrap(err, "increasing bsi bit depth")
 		}
-	}
-
-	// Split import data by fragment.
-	dataByFragment := make(map[importKey]importValueData)
-	for i := range columnIDs {
-		columnID, value := columnIDs[i], values[i]
-		if value > bsig.Max {
-			return fmt.Errorf("%v, columnID=%v, value=%v", ErrBSIGroupValueTooHigh, columnID, value)
-		} else if value < bsig.Min {
-			return fmt.Errorf("%v, columnID=%v, value=%v", ErrBSIGroupValueTooLow, columnID, value)
-		}
-
-		// Attach value to each bsiGroup view.
-		for _, name := range []string{viewName} {
-			key := importKey{View: name, Shard: columnID / ShardWidth}
-			data := dataByFragment[key]
-			data.ColumnIDs = append(data.ColumnIDs, columnID)
-			data.Values = append(data.Values, value)
-			dataByFragment[key] = data
-		}
+	} else {
+		requiredDepth = bsig.BitDepth
 	}
 
 	// Import into each fragment.

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)