From 7a2d90ade8d753bc634d7d76776bf00bf71f0998 Mon Sep 17 00:00:00 2001
From: Todd Gruben <todd@pilosa.com>
Date: Mon, 9 Dec 2019 16:28:03 -0600
Subject: [PATCH] add support to clear value for column

---
 executor.go   | 41 ++++++++++++++++++++++++++++++++++++++++-
 field.go      | 24 +++++++++++++++++++++++-
 field_test.go | 40 ++++++++++++++++++++++++++++++++++++++++
 view.go       | 10 ++++++++++
 4 files changed, 113 insertions(+), 2 deletions(-)

diff --git a/executor.go b/executor.go
index 5d8c94c75..adbd78f41 100644
--- a/executor.go
+++ b/executor.go
@@ -2479,7 +2479,6 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
 	if f == nil {
 		return false, ErrFieldNotFound
 	}
-
 	// Read fields using labels.
 	rowID, ok, err := c.UintArg(fieldName)
 	if err != nil {
@@ -2495,6 +2494,11 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
 		return false, fmt.Errorf("column argument to Clear(<COLUMN>, <FIELD>=<ROW>) required")
 	}
 
+	// Int field.
+	if f.Type() == FieldTypeInt || f.Type() == FieldTypeDecimal {
+		return e.executeClearValueField(ctx, index, c, f, colID, opt)
+	}
+
 	return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
 }
 
@@ -2842,6 +2846,41 @@ func (e *executor) executeSetValueField(ctx context.Context, index string, c *pq
 	return ret, nil
 }
 
+// executeSetValueField executes a Set() call for a specific int field.
+func (e *executor) executeClearValueField(ctx context.Context, index string, c *pql.Call, f *Field, colID uint64, opt *execOptions) (bool, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearValueField")
+	defer span.Finish()
+
+	shard := colID / ShardWidth
+	ret := false
+
+	for _, node := range e.Cluster.shardNodes(index, shard) {
+		// Update locally if host matches.
+		if node.ID == e.Node.ID {
+			val, err := f.ClearValue(colID)
+			if err != nil {
+				return false, err
+			} else if val {
+				ret = true
+			}
+			continue
+		}
+
+		// Do not forward call if this is already being forwarded.
+		if opt.Remote {
+			continue
+		}
+
+		// Forward call to remote node otherwise.
+		res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
+		if err != nil {
+			return false, err
+		}
+		ret = res[0].(bool)
+	}
+	return ret, nil
+}
+
 // executeSetRowAttrs executes a SetRowAttrs() call.
 func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetRowAttrs")
diff --git a/field.go b/field.go
index b6047e3a8..079c0d7d3 100644
--- a/field.go
+++ b/field.go
@@ -1183,10 +1183,32 @@ func (f *Field) SetValue(columnID uint64, value int64) (changed bool, err error)
 	if err != nil {
 		return false, errors.Wrap(err, "creating view")
 	}
-
 	return view.setValue(columnID, bsig.BitDepth, baseValue)
 }
 
+// SetValue sets a field value for a column.
+func (f *Field) ClearValue(columnID uint64) (changed bool, err error) {
+	// Fetch bsiGroup & validate min/max.
+	bsig := f.bsiGroup(f.name)
+	if bsig == nil {
+		return false, ErrBSIGroupNotFound
+	}
+
+	// Fetch target view.
+	view, err := f.createViewIfNotExists(viewBSIGroupPrefix + f.name)
+	if err != nil {
+		return false, errors.Wrap(err, "creating view")
+	}
+	value, exists, err := view.value(columnID, bsig.BitDepth)
+	if err != nil {
+		return false, err
+	}
+	if exists {
+		return view.clearValue(columnID, bsig.BitDepth, value)
+	}
+	return false, nil
+}
+
 // FloatSum performs a Sum query and converts the result to a float
 // based on the field's configured scale.
 func (f *Field) FloatSum(filter *Row, name string) (sum float64, count int64, err error) {
diff --git a/field_test.go b/field_test.go
index 6e7a5d353..79b14e91f 100644
--- a/field_test.go
+++ b/field_test.go
@@ -227,3 +227,43 @@ func TestField_AvailableShards(t *testing.T) {
 		t.Fatal(diff)
 	}
 }
+
+func TestField_ClearValue(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		idx := test.MustOpenIndex()
+		defer idx.Close()
+
+		f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Set value on field.
+		if changed, err := f.SetValue(100, 21); err != nil {
+			t.Fatal(err)
+		} else if !changed {
+			t.Fatal("expected change")
+		}
+
+		// Read value.
+		if value, exists, err := f.Value(100); err != nil {
+			t.Fatal(err)
+		} else if value != 21 {
+			t.Fatalf("unexpected value: %d", value)
+		} else if !exists {
+			t.Fatal("expected value to exist")
+		}
+
+		if changed, err := f.ClearValue(100); err != nil {
+			t.Fatal(err)
+		} else if !changed {
+		}
+
+		// Read value.
+		if _, exists, err := f.Value(100); err != nil {
+			t.Fatal(err)
+		} else if exists {
+			t.Fatal("expected value to not exist")
+		}
+	})
+}
diff --git a/view.go b/view.go
index e1648a838..079a20fac 100644
--- a/view.go
+++ b/view.go
@@ -405,6 +405,16 @@ func (v *view) setValue(columnID uint64, bitDepth uint, value int64) (changed bo
 	return frag.setValue(columnID, bitDepth, value)
 }
 
+// clearValue uses a column of bits to set a multi-bit value.
+func (v *view) clearValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
+	shard := columnID / ShardWidth
+	frag, err := v.CreateFragmentIfNotExists(shard)
+	if err != nil {
+		return changed, err
+	}
+	return frag.clearValue(columnID, bitDepth, value)
+}
+
 // sum returns the sum & count of a field.
 func (v *view) sum(filter *Row, bitDepth uint) (sum int64, count uint64, err error) {
 	for _, f := range v.allFragments() {