Merge pull request #30 from seebs/intfixes

Fix an error that could cause imported values to keep high-order bits from previously imported values, and another that could cause BSI fields to store extra bits they don't need.

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)

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.

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)

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)