[dbnode][coordinator] Support match[] in label endpoints

scripts/docker-integration-tests/prometheus/test.sh

@@ -391,6 +391,47 @@ function test_series {
     '[[ $(curl -s "0.0.0.0:7201/api/v1/series?match[]=prometheus_remote_storage_samples_total&start=-292273086-05-16T16:47:06Z&end=292277025-08-18T07:12:54.999999999Z" | jq -r ".data | length") -eq 1 ]]'
 }
 
+function test_labels {
+  TAG_NAME_0="name_0" TAG_VALUE_0="value_0_1" \
+    TAG_NAME_1="name_1" TAG_VALUE_1="value_1_1" \
+    TAG_NAME_2="name_2" TAG_VALUE_2="value_2_1" \
+    prometheus_remote_write \
+    label_metric now 42.42 \
+    true "Expected request to succeed" \
+    200 "Expected request to return status code 200"
+
+  TAG_NAME_0="name_0" TAG_VALUE_0="value_0_2" \
+    TAG_NAME_1="name_1" TAG_VALUE_1="value_1_2" \
+    prometheus_remote_write \
+    label_metric_2 now 42.42 \
+    true "Expected request to succeed" \
+    200 "Expected request to return status code 200"
+
+  # Test label search with match
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels" | jq -r "[.data[] | select(index(\"name_0\", \"name_1\", \"name_2\"))] | length") -eq 3 ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels?match[]=label_metric" | jq -r ".data | length") -eq 4 ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels?match[]=label_metric_2" | jq -r ".data | length") -eq 3 ]]'
+
+  # Test label values search with match
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values" | jq -r ".data | length") -eq 2 ]]' # two values without a match
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric" | jq -r ".data | length") -eq 1 ]]'
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric" | jq -r ".data[0]") = "value_1_1" ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric_2" | jq -r ".data | length") -eq 1 ]]'
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric_2" | jq -r ".data[0]") = "value_1_2" ]]'
+}
+
 echo "Running readiness test"
 test_readiness
 
@@ -409,6 +450,7 @@ test_prometheus_query_native_timeout
 test_query_restrict_tags
 test_prometheus_remote_write_map_tags
 test_series
+test_labels
 
 echo "Running function correctness tests"
 test_correctness

src/dbnode/storage/index/block.go

@@ -630,7 +630,7 @@ func (b *block) aggregateWithSpan(
 			}
 			return aggOpts.FieldFilter.Allow(field)
 		},
-		fieldIterFn: func(r segment.Reader) (segment.FieldsIterator, error) {
+		fieldIterFn: func(r segment.Reader) (segment.FieldsPostingsListIterator, error) {
 			// NB(prateek): we default to using the regular (FST) fields iterator
 			// unless we have a predefined list of fields we know we need to restrict
 			// our search to, in which case we iterate that list and check if known values
@@ -642,7 +642,7 @@ func (b *block) aggregateWithSpan(
 			// to this function is expected to have (FieldsFilter) pretty small. If that changes
 			// in the future, we can revisit this.
 			if len(aggOpts.FieldFilter) == 0 {
-				return r.Fields()
+				return r.FieldsPostingsList()
 			}
 			return newFilterFieldsIterator(r, aggOpts.FieldFilter)
 		},

src/dbnode/storage/index/fields_terms_iterator.go

@@ -23,11 +23,12 @@ package index
 import (
 	"errors"
 
+	pilosaroaring "github.com/m3dbx/pilosa/roaring"
+
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	"github.com/m3db/m3/src/m3ninx/postings"
 	"github.com/m3db/m3/src/m3ninx/postings/roaring"
 	xerrors "github.com/m3db/m3/src/x/errors"
-	pilosaroaring "github.com/m3dbx/pilosa/roaring"
 )
 
 var (
@@ -49,23 +50,23 @@ func (o fieldsAndTermsIteratorOpts) allow(f []byte) bool {
 	return o.allowFn(f)
 }
 
-func (o fieldsAndTermsIteratorOpts) newFieldIter(r segment.Reader) (segment.FieldsIterator, error) {
+func (o fieldsAndTermsIteratorOpts) newFieldIter(r segment.Reader) (segment.FieldsPostingsListIterator, error) {
 	if o.fieldIterFn == nil {
-		return r.Fields()
+		return r.FieldsPostingsList()
 	}
 	return o.fieldIterFn(r)
 }
 
 type allowFn func(field []byte) bool
 
-type newFieldIterFn func(r segment.Reader) (segment.FieldsIterator, error)
+type newFieldIterFn func(r segment.Reader) (segment.FieldsPostingsListIterator, error)
 
 type fieldsAndTermsIter struct {
 	reader segment.Reader
 	opts   fieldsAndTermsIteratorOpts
 
 	err       error
-	fieldIter segment.FieldsIterator
+	fieldIter segment.FieldsPostingsListIterator
 	termIter  segment.TermsIterator
 
 	current struct {
@@ -145,10 +146,33 @@ func (fti *fieldsAndTermsIter) setNextField() bool {
 	}
 
 	for fieldIter.Next() {
-		field := fieldIter.Current()
+		field, pl := fieldIter.Current()
 		if !fti.opts.allow(field) {
 			continue
 		}
+		if fti.restrictByPostings == nil {
+			// No restrictions.
+			fti.current.field = field
+			return true
+		}
+
+		bitmap, ok := roaring.BitmapFromPostingsList(pl)
+		if !ok {
+			fti.err = errUnpackBitmapFromPostingsList
+			return false
+		}
+
+		// Check field is part of at least some of the documents we're
+		// restricted to providing results for based on intersection
+		// count.
+		// Note: IntersectionCount is significantly faster than intersecting and
+		// counting results and also does not allocate.
+		if n := fti.restrictByPostings.IntersectionCount(bitmap); n < 1 {
+			// No match, not next result.
+			continue
+		}
+
+		// Matches, this is next result.
 		fti.current.field = field
 		return true
 	}
@@ -213,7 +237,7 @@ func (fti *fieldsAndTermsIter) nextTermsIterResult() (bool, error) {
 			return false, errUnpackBitmapFromPostingsList
 		}
 
-		// Check term isn part of at least some of the documents we're
+		// Check term is part of at least some of the documents we're
 		// restricted to providing results for based on intersection
 		// count.
 		// Note: IntersectionCount is significantly faster than intersecting and

src/dbnode/storage/index/fields_terms_iterator_prop_test.go

@@ -35,6 +35,8 @@ import (
 	"github.com/leanovate/gopter"
 	"github.com/leanovate/gopter/gen"
 	"github.com/leanovate/gopter/prop"
+	"github.com/stretchr/testify/require"
+
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	xtest "github.com/m3db/m3/src/x/test"
 )
@@ -62,7 +64,8 @@ func TestFieldsTermsIteratorPropertyTest(t *testing.T) {
 			if err != nil {
 				return false, err
 			}
-			observed := toSlice(t, iter)
+			observed, err := toSlice(iter)
+			require.NoError(t, err)
 			requireSlicesEqual(t, expected, observed)
 			return true, nil
 		},
@@ -97,7 +100,7 @@ func TestFieldsTermsIteratorPropertyTestNoPanic(t *testing.T) {
 			if err != nil {
 				return false, err
 			}
-			toSlice(t, iter)
+			_, _ = toSlice(iter)
 			return true, nil
 		},
 		genIterableSegment(ctrl),
@@ -149,9 +152,9 @@ func genIterableSegment(ctrl *gomock.Controller) gopter.Gen {
 
 			r := segment.NewMockReader(ctrl)
 
-			fieldIterator := &stubFieldIterator{points: fields}
+			fieldsPostingsListIterator := &stubFieldsPostingsListIterator{points: fields}
 
-			r.EXPECT().Fields().Return(fieldIterator, nil).AnyTimes()
+			r.EXPECT().FieldsPostingsList().Return(fieldsPostingsListIterator, nil).AnyTimes()
 
 			for f, values := range tagValues {
 				sort.Slice(values, func(i, j int) bool {