Repository: pilosa/pilosa (go)

Issue:
import to integer field seems to result in incorrect values
### What's going wrong?
Integer values imported with `pilosa import` seem to be getting set incorrectly when they come after a `Row()` query. I haven't dug into this, but I wanted to report it as it seems to be a problem. See the example below for what's happening.

### What was expected?
Separate imports result in the same values as a single import.

### Steps to reproduce the behavior

This works as expected:
```
curl localhost:10101/index/t -d '{"options": {"keys": false, "trackExistence": false}}'
curl localhost:10101/index/t/field/f -d '{"options": {"type": "int", "keys": false}}'
echo "2,1" > tmp.csv; pilosa import -i t -f f tmp.csv
echo "1000,1" > tmp.csv; pilosa import -i t -f f tmp.csv
curl localhost:10101/index/t/query -d 'Row(f>0)'
```
The result is:
```
{"results":[{"attrs":{},"columns":[2,1000]}]}
```

However, when a `Row` query is run between the two imports, the second import seems to get applied incorrectly.
```
curl localhost:10101/index/t2 -d '{"options": {"keys": false, "trackExistence": false}}'
curl localhost:10101/index/t2/field/f2 -d '{"options": {"type": "int", "keys": false}}'
echo "2,1" > tmp.csv; pilosa import -i t2 -f f2 tmp.csv
curl localhost:10101/index/t2/query -d 'Row(f2>0)'
echo "1000,1" > tmp.csv; pilosa import -i t2 -f f2 tmp.csv
curl localhost:10101/index/t2/query -d 'Row(f2>0)'
curl localhost:10101/index/t2/query -d 'Row(f2<0)'
```
The result here is:
```
{"results":[{"attrs":{},"columns":[2]}]}
{"results":[{"attrs":{},"columns":[1000]}]}
```

### Information about your environment (OS/architecture, CPU, RAM, cluster/solo, configuration, etc.)


For feature requests, please provide the following:

### Description

### Success criteria (What criteria will consider this ticket closeable?)


Pull request description:
Add logic that clears fragment.rowCache after importValue
## Overview

It seems that `fragment.importValue` doesn't clear `fragment.rowCache` the way that other mutating methods do (see: `fragment.importRoaring`, `fragment.unprotectedSetBit`, etc.).

This PR is pretty heavy-handed in that it clears the rowCache for all rows of the BSI value, as well as the existence and sign rows. There's likely a more elegant way to handle this, but I'm putting this here to get get feedback.

TODO:
- [x] If we can settle on the appropriate solution for this, I will add tests at the `fragment` level (currently, the test that exercises this is in the `http` package).

Fixes #2083 

## Pull request checklist

- [x] I have read the [contributing guide](https://github.com/pilosa/pilosa/blob/master/CONTRIBUTING.md).
- [x] I have agreed to the [Contributor License Agreement](https://cla-assistant.io/pilosa/pilosa).
- [ ] I have updated the [documentation](https://github.com/pilosa/pilosa/tree/master/docs).
- [x] I have resolved any merge conflicts.
- [x] I have included tests that cover my changes.
- [x] All new and existing tests pass.
- [x] Make sure PR title conforms to convention in CHANGELOG.md.
- [x] Add appropriate changelog label to PR (if applicable).

## Code review checklist
This is the checklist that the reviewer will follow while reviewing your pull request. You do not need to do anything with this checklist, but be aware of what the reviewer will be looking for.

- [ ] Ensure that any changes to external docs have been included in this pull request.
- [ ] If the changes require that minor/major versions need to be updated, tag the PR appropriately.
- [ ] Ensure the new code is [properly commented](https://github.com/golang/go/wiki/CodeReviewComments#doc-comments) and follows [Idiomatic Go](https://dmitri.shuralyov.com/idiomatic-go).
- [ ] Check that tests have been written and that they cover the new functionality.
- [ ] Run tests and ensure they pass.
- [ ] Build and run the code, performing any applicable integration testing.
- [ ] Make sure PR title conforms to convention in CHANGELOG.md.
- [ ] Make sure PR is tagged with appropriate changelog label.


Patch:
diff --git a/field.go b/field.go
index 38b364b6..fe7c0255 100644
--- a/field.go
+++ b/field.go
@@ -1292,15 +1292,38 @@ 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.
@@ -1308,7 +1331,6 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
 	if v := bitDepthInt64(max - bsig.Base); v > requiredDepth {
 		requiredDepth = v
 	}
-
 	// Increase bit depth if required.
 	if requiredDepth > bsig.BitDepth {
 		if err := func() error {
@@ -1320,26 +1342,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.
diff --git a/fragment.go b/fragment.go
index 03fc11b6..1690e5e3 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.


Test patch:
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index 1181d5e8..92b6d107 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 d61640d7..0b655851 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)