Merge pull request #62 from tgruben/clearvalue

add support to clear value for column

executor.go

@@ -2650,6 +2650,14 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearBit")
 	defer span.Finish()
 
+	// Read colID
+	colID, ok, err := c.UintArg("_" + columnLabel)
+	if err != nil {
+		return false, fmt.Errorf("reading Clear() column: %v", err)
+	} else if !ok {
+		return false, fmt.Errorf("column argument to Clear(<COLUMN>, <FIELD>=<ROW>) required")
+	}
+	// Read field name.
 	fieldName, err := c.FieldArg()
 	if err != nil {
 		return false, errors.New("Clear() argument required: field")
@@ -2665,7 +2673,18 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
 		return false, ErrFieldNotFound
 	}
 
-	// Read fields using labels.
+	// Clear column on existence field
+	if ef := idx.existenceField(); ef != nil {
+		if _, err := ef.ClearBit(0, colID); err != nil {
+			return false, errors.Wrap(err, "clearing existence column")
+		}
+	}
+
+	// Int field.
+	if f.Type() == FieldTypeInt || f.Type() == FieldTypeDecimal {
+		return e.executeClearValueField(ctx, index, c, f, colID, opt)
+	}
+
 	rowID, ok, err := c.UintArg(fieldName)
 	if err != nil {
 		return false, fmt.Errorf("reading Clear() row: %v", err)
@@ -2673,13 +2692,6 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
 		return false, fmt.Errorf("row=<row> argument required to Clear() call")
 	}
 
-	colID, ok, err := c.UintArg("_" + columnLabel)
-	if err != nil {
-		return false, fmt.Errorf("reading Clear() column: %v", err)
-	} else if !ok {
-		return false, fmt.Errorf("column argument to Clear(<COLUMN>, <FIELD>=<ROW>) required")
-	}
-
 	return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
 }
 
@@ -3027,6 +3039,41 @@ func (e *executor) executeSetValueField(ctx context.Context, index string, c *pq
 	return ret, nil
 }
 
+// executeClearValueField removes value for colID if present
+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")

field.go

@@ -1183,10 +1183,30 @@ 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)
 }
 
+// ClearValue removes a field value for a column.
+func (f *Field) ClearValue(columnID uint64) (changed bool, err error) {
+	bsig := f.bsiGroup(f.name)
+	if bsig == nil {
+		return false, ErrBSIGroupNotFound
+	}
+	// Fetch target view.
+	view := f.view(viewBSIGroupPrefix + f.name)
+	if view == nil {
+		return false, nil
+	}
+	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) {

field_test.go

@@ -227,3 +227,44 @@ 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 {
+			t.Fatal(err)
+		}
+
+		// Read value.
+		if _, exists, err := f.Value(100); err != nil {
+			t.Fatal(err)
+		} else if exists {
+			t.Fatal("expected value to not exist")
+		}
+	})
+}

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 removes a specific value assigned to columnID
+func (v *view) clearValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
+	shard := columnID / ShardWidth
+	frag := v.Fragment(shard)
+	if frag == nil {
+		return false, nil
+	}
+	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() {