From 5ebd7e8da8b10bf91f380d5bdbae669212488e3a Mon Sep 17 00:00:00 2001
From: Travis Turner <travis@pilosa.com>
Date: Mon, 11 Nov 2019 16:48:14 -0600
Subject: [PATCH] move requiredDepth calculation after min/max ranges are
 checked

Without this, a data set with a ludicrously large value in it
could break a BSI field's depth even though the import would then
reject it.

always treat BSI fields as having at least their depth:

If you imported only small values, BSI fields could end up
not bothering to clear higher bits in existing values, which
produced strange behaviors.
---
 field.go | 58 ++++++++++++++++++++++++++++++--------------------------
 1 file changed, 31 insertions(+), 27 deletions(-)

diff --git a/field.go b/field.go
index 39c6c5363..a4b36e8d4 100644
--- a/field.go
+++ b/field.go
@@ -1269,34 +1269,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.
@@ -1308,6 +1286,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} {
@@ -1319,6 +1303,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)