[dbnode] Allow returning empty term values from aggregate index value…

…s RPC (#2120)

src/dbnode/storage/index/aggregate_results.go

@@ -238,10 +238,6 @@ func (r *aggregatedResults) addFieldWithLock(
 		return fmt.Errorf(missingDocumentFields, "term")
 	}
 
-	if len(value) == 0 {
-		return fmt.Errorf(missingDocumentFields, "value")
-	}
-
 	// if a term filter is provided, ensure this field matches the filter,
 	// otherwise ignore it.
 	filter := r.aggregateOpts.FieldFilter

src/dbnode/storage/index/aggregate_results_test.go

@@ -59,11 +59,14 @@ func TestAggResultsInsertInvalid(t *testing.T) {
 	size, err = res.AddDocuments([]doc.Document{dInvalid})
 	require.Error(t, err)
 	require.Equal(t, 0, size)
+}
 
-	dInvalid = genDoc("foo", "")
-	size, err = res.AddDocuments([]doc.Document{dInvalid})
-	require.Error(t, err)
-	require.Equal(t, 0, size)
+func TestAggResultsInsertEmptyTermValue(t *testing.T) {
+	res := NewAggregateResults(nil, AggregateResultsOptions{}, testOpts)
+	dValidEmptyTerm := genDoc("foo", "")
+	size, err := res.AddDocuments([]doc.Document{dValidEmptyTerm})
+	require.NoError(t, err)
+	require.Equal(t, 1, size)
 }
 
 func TestAggResultsTermOnlyInsert(t *testing.T) {

src/dbnode/storage/index/aggregate_values.go

@@ -21,16 +21,10 @@
 package index
 
 import (
-	"errors"
-
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/pool"
 )
 
-var (
-	errUnableToAddValueMissingID = errors.New("no id for value")
-)
-
 // AggregateValues is a collection of unique identity values backed by a pool.
 // NB: there are no synchronization guarantees provided by default.
 type AggregateValues struct {
@@ -83,10 +77,9 @@ func (v *AggregateValues) finalize() {
 }
 
 func (v *AggregateValues) addValue(value ident.ID) error {
+	// NB(r): Allow for empty values to be set, an empty string
+	// is still different from not having the field at all.
 	bytesID := ident.BytesID(value.Bytes())
-	if len(bytesID) == 0 {
-		return errUnableToAddValueMissingID
-	}
 
 	// NB: fine to overwrite the values here.
 	v.valuesMap.Set(bytesID, struct{}{})