Merge pull request #16530 from influxdata/er-fix-predicate

fix(storage): prevent infinite loop in predicate matcher

tsdb/tsm1/predicate.go

@@ -621,17 +621,16 @@ func predicatePopTagEscape(series []byte) (tag, value []byte, rest []byte) {
 	for j := uint(0); j < uint(len(series)); {
 		i := bytes.IndexByte(series[j:], ',')
 		if i < 0 {
-			break
+			break // this is the last tag pair
 		}
-		ui := uint(i)
 
-		if ui > 0 && ui-1 < uint(len(series)) && series[ui-1] == '\\' {
+		ui := uint(i) + j                   // make index relative to full series slice
+		if ui > 0 && series[ui-1] == '\\' { // the comma is escaped
 			j = ui + 1
 			continue
 		}
 
-		idx := ui + j
-		series, rest = series[:idx], series[idx+1:]
+		series, rest = series[:ui], series[ui+1:]
 		break
 	}
 
@@ -639,17 +638,15 @@ func predicatePopTagEscape(series []byte) (tag, value []byte, rest []byte) {
 	for j := uint(0); j < uint(len(series)); {
 		i := bytes.IndexByte(series[j:], '=')
 		if i < 0 {
-			break
+			break // there is no tag value
 		}
-		ui := uint(i)
-
-		if ui > 0 && ui-1 < uint(len(series)) && series[ui-1] == '\\' {
+		ui := uint(i) + j                   // make index relative to full series slice
+		if ui > 0 && series[ui-1] == '\\' { // the equals is escaped
 			j = ui + 1
 			continue
 		}
 
-		idx := ui + j
-		tag, value = series[:idx], series[idx+1:]
+		tag, value = series[:ui], series[ui+1:]
 		break
 	}
 

tsdb/tsm1/predicate_test.go

@@ -8,6 +8,35 @@ import (
 	"github.com/influxdata/influxdb/storage/reads/datatypes"
 )
 
+func TestPredicatePopTagEscape(t *testing.T) {
+	cases := []struct {
+		Key   string
+		Tag   string
+		Value string
+		Rest  string
+	}{
+		{Key: "", Tag: "", Value: "", Rest: ""},
+		{Key: "invalid", Tag: "", Value: "", Rest: ""},
+		{Key: "region=west,server=b,foo=bar", Tag: "region", Value: "west", Rest: "server=b,foo=bar"},
+		{Key: "region=west", Tag: "region", Value: "west", Rest: ""},
+		{Key: `re\=gion=west,server=a`, Tag: `re=gion`, Value: "west", Rest: "server=a"},
+		{Key: `region=w\,est,server=a`, Tag: `region`, Value: "w,est", Rest: "server=a"},
+		{Key: `hi\ yo\ =w\,est,server=a`, Tag: `hi yo `, Value: "w,est", Rest: "server=a"},
+		{Key: `\ e\ \=o=world,server=a`, Tag: ` e =o`, Value: "world", Rest: "server=a"},
+	}
+
+	for _, c := range cases {
+		tag, value, rest := predicatePopTagEscape([]byte(c.Key))
+		if string(tag) != c.Tag {
+			t.Fatalf("got returned tag %q expected %q", tag, c.Tag)
+		} else if string(value) != c.Value {
+			t.Fatalf("got returned value %q expected %q", value, c.Value)
+		} else if string(rest) != c.Rest {
+			t.Fatalf("got returned remainder %q expected %q", rest, c.Rest)
+		}
+	}
+}
+
 func TestPredicate_Matches(t *testing.T) {
 	cases := []struct {
 		Name      string