Add parsing of the indirect access operator ('=>').

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/adrg/xdg v0.5.3
 	github.com/golang/protobuf v1.5.4
 	github.com/google/gnostic-models v0.6.9
-	github.com/google/go-cmp v0.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.3
 	github.com/pborman/uuid v1.2.1
@@ -80,6 +79,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.0.1 // indirect
 	github.com/google/cel-go v0.22.0 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect

pkg/sqlcache/sqltypes/types.go

@@ -44,6 +44,8 @@ type Filter struct {
 	Matches        []string
 	Op             Op
 	Partial        bool
+	IsIndirect     bool
+	IndirectFields []string
 }
 
 // OrFilter represents a set of possible fields to filter by, where an item may match any filter in the set to be included in the result.
@@ -59,10 +61,8 @@ type OrFilter struct {
 type Sort struct {
 	Fields         []string
 	Order          SortOrder
-}
+	IsIndirect     bool
-
+	IndirectFields []string
-type SortList struct {
-	SortDirectives []Sort
 }
 
 type SortList struct {
@@ -80,9 +80,3 @@ func NewSortList() *SortList {
 		SortDirectives: []Sort{},
 	}
 }
-
-func NewSortList() *SortList {
-	return &SortList{
-		SortDirectives: []Sort{},
-	}
-}

pkg/stores/sqlpartition/listprocessor/processor.go

@@ -3,6 +3,7 @@ package listprocessor
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -93,7 +94,7 @@ func ParseQuery(apiOp *types.APIRequest, namespaceCache Cache) (sqltypes.ListOpt
 	filterParams := q[filterParam]
 	filterOpts := []sqltypes.OrFilter{}
 	for _, filters := range filterParams {
-		requirements, err := queryparser.ParseToRequirements(filters)
+		requirements, err := queryparser.ParseToRequirements(filters, filterParam)
 		if err != nil {
 			return sqltypes.ListOptions{}, err
 		}
@@ -109,27 +110,38 @@ func ParseQuery(apiOp *types.APIRequest, namespaceCache Cache) (sqltypes.ListOpt
 	}
 	opts.Filters = filterOpts
 
+	if q.Has(sortParam) {
 		sortKeys := q.Get(sortParam)
-	if sortKeys != "" {
+		filterRequirements, err := queryparser.ParseToRequirements(sortKeys, sortParam)
-		sortList := *sqltypes.NewSortList()
+		if err != nil {
-		sortParts := strings.Split(sortKeys, ",")
+			return opts, err
-		for _, sortPart := range sortParts {
-			field := sortPart
-			if len(field) > 0 {
-				sortOrder := sqltypes.ASC
-				if field[0] == '-' {
-					sortOrder = sqltypes.DESC
-					field = field[1:]
 		}
-				if len(field) > 0 {
+		if len(filterRequirements) == 0 {
+			if len(sortKeys) == 0 {
+				return opts, errors.New("invalid sort key: <empty string>")
+			}
+			return opts, fmt.Errorf("invalid sort key: '%s'", sortKeys)
+		}
+		sortList := *sqltypes.NewSortList()
+		for _, requirement := range filterRequirements {
+			if requirement.Operator() != selection.Exists {
+				return opts, fmt.Errorf("sort directive %s can't contain operator (%s)", sortKeys, requirement.Operator())
+			}
+			key := requirement.Key()
+			order := sqltypes.ASC
+			if key[0] == '-' {
+				order = sqltypes.DESC
+				key = key[1:]
+			}
+			isIndirect, indirectFields := requirement.IndirectInfo()
 			sortDirective := sqltypes.Sort{
-						Fields: queryhelper.SafeSplit(field),
+				Fields:         queryhelper.SafeSplit(key),
-						Order:  sortOrder,
+				Order:          order,
+				IsIndirect:     isIndirect,
+				IndirectFields: indirectFields,
 			}
 			sortList.SortDirectives = append(sortList.SortDirectives, sortDirective)
 		}
-			}
-		}
 		opts.SortList = sortList
 	}
 

pkg/stores/sqlpartition/queryparser/selector.go

@@ -39,6 +39,9 @@ the array into a sql statement. So the set gives us no benefit apart from removi
 6.  We allow `lt` and `gt` as aliases for `<` and `>`.
 
 7. We added the '~' and '!~' operators to indicate partial match and non-match
+
+8. We added indirect field selection so we can base a filter or sort off a related value
+  (could be in a different table)
 */
 
 package queryparser
@@ -70,6 +73,7 @@ var (
 		string(selection.Equals), string(selection.DoubleEquals), string(selection.NotEquals),
 		string(selection.PartialEquals), string(selection.NotPartialEquals),
 		string(selection.GreaterThan), string(selection.LessThan),
+		string(selection.IndirectSelector),
 	}
 	validRequirementOperators = append(binaryOperators, unaryOperators...)
 	labelSelectorRegex        = regexp.MustCompile(`^metadata.labels(?:\.\w[-a-zA-Z0-9_./]*|\[.*])$`)
@@ -136,6 +140,8 @@ type Requirement struct {
 	// It is generally faster to operate on a single-element slice
 	// than on a single-element map, so we have a slice here.
 	strValues      []string
+	isIndirect     bool
+	indirectFields []string
 }
 
 // NewRequirement is the constructor for a Requirement.
@@ -183,7 +189,26 @@ func NewRequirement(key string, op selection.Operator, vals []string, opts ...fi
 	default:
 		allErrs = append(allErrs, field.NotSupported(path.Child("operator"), op, validRequirementOperators))
 	}
-	return &Requirement{key: key, operator: op, strValues: vals}, allErrs.ToAggregate()
+	agg := allErrs.ToAggregate()
+	var err error
+	if agg != nil {
+		err = errors.New(agg.Error())
+	}
+	return &Requirement{key: key, operator: op, strValues: vals}, err
+}
+
+func NewIndirectRequirement(key string, indirectFields []string, newOperator *selection.Operator, targetValues []string, opts ...field.PathOption) (*Requirement, error) {
+	if newOperator == nil {
+		operator := selection.Exists
+		newOperator = &operator
+	}
+	r, err := NewRequirement(key, *newOperator, targetValues)
+	if err != nil {
+		return nil, err
+	}
+	r.isIndirect = true
+	r.indirectFields = indirectFields
+	return r, nil
 }
 
 func (r *Requirement) hasValue(value string) bool {
@@ -214,6 +239,10 @@ func (r *Requirement) Values() []string {
 	return ret.List()
 }
 
+func (r *Requirement) IndirectInfo() (bool, []string) {
+	return r.isIndirect, r.indirectFields
+}
+
 // Equal checks the equality of requirement.
 func (r Requirement) Equal(x Requirement) bool {
 	if r.key != x.key {
@@ -377,6 +406,8 @@ const (
 	NotPartialEqualsToken
 	// OpenParToken represents open parenthesis
 	OpenParToken
+	// IndirectAccessToken is =>, used to associate one table with a related one, and grab a different field
+	IndirectAccessToken
 )
 
 // string2token contains the mapping between lexer Token and token literal
@@ -395,6 +426,7 @@ var string2token = map[string]Token{
 	"!~":    NotPartialEqualsToken,
 	"notin": NotInToken,
 	"(":     OpenParToken,
+	"=>":    IndirectAccessToken,
 }
 
 // ScannedItem contains the Token and the literal produced by the lexer.
@@ -405,7 +437,7 @@ type ScannedItem struct {
 
 func isIdentifierStartChar(ch byte) bool {
 	r := rune(ch)
-	return unicode.IsLetter(r) || unicode.IsDigit(r) || ch == '_'
+	return unicode.IsLetter(r) || unicode.IsDigit(r) || ch == '_' || ch == '[' || ch == '-'
 }
 
 // isWhitespace returns true if the rune is a space, tab, or newline.
@@ -531,6 +563,7 @@ type Parser struct {
 	l            *Lexer
 	scannedItems []ScannedItem
 	position     int
+	parseType    string
 	path         *field.Path
 }
 
@@ -624,13 +657,18 @@ func (p *Parser) parseRequirement() (*Requirement, error) {
 	if err != nil {
 		return nil, err
 	}
+	fieldPath := field.WithPath(p.path)
 	if operator == selection.Exists || operator == selection.DoesNotExist { // operator found lookahead set checked
-		if !labelSelectorRegex.MatchString(key) {
+		if p.parseType == "filter" && !labelSelectorRegex.MatchString(key) {
 			return nil, fmt.Errorf("existence tests are valid only for labels; not valid for field '%s'", key)
 		}
-		return NewRequirement(key, operator, []string{}, field.WithPath(p.path))
+		return NewRequirement(key, operator, []string{}, fieldPath)
 	}
-	operator, err = p.parseOperator()
+	return p.parseOperatorAndValues(key, fieldPath, true)
+}
+
+func (p *Parser) parseOperatorAndValues(key string, fieldPath field.PathOption, allowIndirectSelector bool) (*Requirement, error) {
+	operator, err := p.parseOperator()
 	if err != nil {
 		return nil, err
 	}
@@ -640,12 +678,22 @@ func (p *Parser) parseRequirement() (*Requirement, error) {
 		values, err = p.parseValues()
 	case selection.Equals, selection.DoubleEquals, selection.NotEquals, selection.GreaterThan, selection.LessThan, selection.PartialEquals, selection.NotPartialEquals:
 		values, err = p.parseSingleValue()
+	case selection.IndirectSelector:
+		if !allowIndirectSelector {
+			return nil, fmt.Errorf("found a subsequent indirect selector (->)")
+		}
+		indirectFields, newOperator, targetValues, err := p.parseIndirectAccessorPart(key, fieldPath)
+		if err != nil {
+			return nil, err
+		} else if newOperator != nil && p.parseType == "sort" {
+			return nil, fmt.Errorf("found an operator (%s) in a sort expression )", *newOperator)
+		}
+		return NewIndirectRequirement(key, indirectFields, newOperator, targetValues.List(), fieldPath)
 	}
 	if err != nil {
 		return nil, err
 	}
 	return NewRequirement(key, operator, values.List(), field.WithPath(p.path))
-
 }
 
 // parseKeyAndInferOperator parses literals.
@@ -694,11 +742,15 @@ func (p *Parser) parseOperator() (op selection.Operator, err error) {
 		op = selection.NotEquals
 	case NotPartialEqualsToken:
 		op = selection.NotPartialEquals
+	case IndirectAccessToken:
+		op = selection.IndirectSelector
 	default:
 		if lit == "lt" {
 			op = selection.LessThan
 		} else if lit == "gt" {
 			op = selection.GreaterThan
+		} else if p.parseType == "sort" {
+			return "", fmt.Errorf("found unexpected token '%s' in sort parameter", lit)
 		} else {
 			return "", fmt.Errorf("found '%s', expected: %v", lit, strings.Join(binaryOperators, ", "))
 		}
@@ -727,10 +779,38 @@ func (p *Parser) parseValues() (sets.String, error) {
 		p.consume(Values)
 		return sets.NewString(""), nil
 	default:
-		return nil, fmt.Errorf("found '%s', expected: ',', ')' or identifier", lit)
+		return sets.NewString(""), fmt.Errorf("found '%s', expected: ',', ')' or identifier", lit)
 	}
 }
 
+func (p *Parser) parseIndirectAccessorPart(key string, fieldPath field.PathOption) ([]string, *selection.Operator, sets.String, error) {
+	//key string, indirectFields []string, newOperator selection.Operator, targetValues []string
+	values := sets.String{}
+	tok, lit := p.consume(Values)
+	if tok != IdentifierToken {
+		return nil, nil, values, fmt.Errorf("found '%s', expected: an indirect field specifier", lit)
+	}
+	matched, err := regexp.MatchString(`^(?:\[.*?\])+$`, lit)
+	if err != nil {
+		return nil, nil, values, err
+	} else if !matched {
+		return nil, nil, values, fmt.Errorf("found '%s', expected: a sequence of bracketed identifiers", lit)
+	}
+
+	indirectFields := strings.Split(lit[1:len(lit)-1], "][")
+	if len(indirectFields) != 4 {
+		return nil, nil, values, fmt.Errorf("found '%s', expected: a sequence of three bracketed identifiers", lit)
+	}
+	if p.parseType == "sort" {
+		return indirectFields, nil, sets.NewString(), nil
+	}
+	r, err := p.parseOperatorAndValues(key, fieldPath, false)
+	if err != nil {
+		return nil, nil, values, err
+	}
+	return indirectFields, &r.operator, sets.NewString(r.strValues...), nil
+}
+
 // parseIdentifiersList parses a (possibly empty) list of
 // of comma separated (possibly empty) identifiers
 func (p *Parser) parseIdentifiersList() (sets.String, error) {
@@ -814,9 +894,9 @@ func (p *Parser) parseSingleValue() (sets.String, error) {
 //  4. A requirement with just a KEY - as in "y" above - denotes that
 //     the KEY exists and can be any VALUE.
 //  5. A requirement with just !KEY requires that the KEY not exist.
-func Parse(selector string, opts ...field.PathOption) (Selector, error) {
+func Parse(selector string, parseType string, opts ...field.PathOption) (Selector, error) {
 	pathThing := field.ToPath(opts...)
-	parsedSelector, err := parse(selector, pathThing)
+	parsedSelector, err := parse(selector, parseType, pathThing)
 	if err == nil {
 		return parsedSelector, nil
 	}
@@ -827,8 +907,8 @@ func Parse(selector string, opts ...field.PathOption) (Selector, error) {
 // The callers of this method can then decide how to return the internalSelector struct to their
 // callers. This function has two callers now, one returns a Selector interface and the other
 // returns a list of requirements.
-func parse(selector string, path *field.Path) (internalSelector, error) {
+func parse(selector string, parseType string, path *field.Path) (internalSelector, error) {
-	p := &Parser{l: &Lexer{s: selector, pos: 0}, path: path}
+	p := &Parser{l: &Lexer{s: selector, pos: 0}, parseType: parseType, path: path}
 	items, err := p.parse()
 	if err != nil {
 		return nil, err
@@ -883,8 +963,8 @@ func SelectorFromValidatedSet(ls Set) Selector {
 // processing on selector requirements.
 // See the documentation for Parse() function for more details.
 // TODO: Consider exporting the internalSelector type instead.
-func ParseToRequirements(selector string, opts ...field.PathOption) ([]Requirement, error) {
+func ParseToRequirements(selector string, parseType string, opts ...field.PathOption) ([]Requirement, error) {
-	return parse(selector, field.ToPath(opts...))
+	return parse(selector, parseType, field.ToPath(opts...))
 }
 
 // ValidatedSetSelector wraps a Set, allowing it to implement the Selector interface. Unlike

pkg/stores/sqlpartition/queryparser/selector_test.go

@@ -27,15 +27,9 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/rancher/steve/pkg/stores/sqlpartition/selection"
+	"github.com/stretchr/testify/assert"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/util/validation/field"
-)
-
-var (
-	ignoreDetail = cmpopts.IgnoreFields(field.Error{}, "Detail")
 )
 
 func TestSelectorParse(t *testing.T) {
@@ -54,6 +48,9 @@ func TestSelectorParse(t *testing.T) {
 		"metadata.labels[im.here]",
 		"!metadata.labels[im.not.here]",
 		"metadata.labels[k8s.io/meta-stuff] ~ has-dashes_underscores.dots.only",
+		"metadata.labels[k8s.io/meta-stuff] => [management.cattle.io/v3][tokens][id][metadata.state.name] = active",
+		"name => [management.cattle.io/v3][tokens][id][metadata.state.name] = active",
+		"metadata.annotations[blah] => [management.cattle.io/v3][tokens][id][metadata.state.name] = active",
 	}
 	testBadStrings := []string{
 		"!no-label-absence-test",
@@ -77,15 +74,22 @@ func TestSelectorParse(t *testing.T) {
 		"!metadata.labels(im.not.here)",
 		`x="no double quotes allowed"`,
 		`x='no single quotes allowed'`,
+		"metadata.labels[k8s.io/meta-stuff] => not-bracketed = active",
+		"metadata.labels[k8s.io/meta-stuff] => [not][enough][accessors] = active",
+		"metadata.labels[k8s.io/meta-stuff] => [too][many][accessors][by][1] = active",
+		"metadata.labels[k8s.io/meta-stuff] => [missing][an][operator][end-of-string]",
+		"metadata.labels[k8s.io/meta-stuff] => [missing][an][operator][no-following-operator] no-operator",
+		"metadata.labels[k8s.io/meta-stuff] => [missing][a][post-operator][value] >",
+		"metadata.labels[not/followed/by/accessor] => = active",
 	}
 	for _, test := range testGoodStrings {
-		_, err := Parse(test)
+		_, err := Parse(test, "filter")
 		if err != nil {
 			t.Errorf("%v: error %v (%#v)\n", test, err, err)
 		}
 	}
 	for _, test := range testBadStrings {
-		_, err := Parse(test)
+		_, err := Parse(test, "filter")
 		if err == nil {
 			t.Errorf("%v: did not get expected error\n", test)
 		}
@@ -115,6 +119,7 @@ func TestLexer(t *testing.T) {
 		{"~", PartialEqualsToken},
 		{"!~", NotPartialEqualsToken},
 		{"||", ErrorToken},
+		{"=>", IndirectAccessToken},
 	}
 	for _, v := range testcases {
 		l := &Lexer{s: v.s, pos: 0}
@@ -163,6 +168,9 @@ func TestLexerSequence(t *testing.T) {
 		{"key!~ value", []Token{IdentifierToken, NotPartialEqualsToken, IdentifierToken}},
 		{"key !~value", []Token{IdentifierToken, NotPartialEqualsToken, IdentifierToken}},
 		{"key!~value", []Token{IdentifierToken, NotPartialEqualsToken, IdentifierToken}},
+		{"metadata.labels[k8s.io/meta-stuff] => [management.cattle.io/v3][tokens][id][metadata.state.name] = active",
+			[]Token{IdentifierToken, IndirectAccessToken, IdentifierToken, EqualsToken, IdentifierToken},
+		},
 	}
 	for _, v := range testcases {
 		var tokens []Token
@@ -203,6 +211,10 @@ func TestParserLookahead(t *testing.T) {
 		{"key gt 3", []Token{IdentifierToken, GreaterThanToken, IdentifierToken, EndOfStringToken}},
 		{"key lt 4", []Token{IdentifierToken, LessThanToken, IdentifierToken, EndOfStringToken}},
 		{`key = multi-word-string`, []Token{IdentifierToken, EqualsToken, QuotedStringToken, EndOfStringToken}},
+
+		{"metadata.labels[k8s.io/meta-stuff] => [management.cattle.io/v3][tokens][id][metadata.state.name] = active",
+			[]Token{IdentifierToken, IndirectAccessToken, IdentifierToken, EqualsToken, IdentifierToken, EndOfStringToken},
+		},
 	}
 	for _, v := range testcases {
 		p := &Parser{l: &Lexer{s: v.s, pos: 0}, position: 0}
@@ -240,6 +252,7 @@ func TestParseOperator(t *testing.T) {
 		{"notin", nil},
 		{"!=", nil},
 		{"!~", nil},
+		{"=>", nil},
 		{"!", fmt.Errorf("found '%s', expected: %v", selection.DoesNotExist, strings.Join(binaryOperators, ", "))},
 		{"exists", fmt.Errorf("found '%s', expected: %v", selection.Exists, strings.Join(binaryOperators, ", "))},
 		{"(", fmt.Errorf("found '%s', expected: %v", "(", strings.Join(binaryOperators, ", "))},
@@ -262,30 +275,18 @@ func TestRequirementConstructor(t *testing.T) {
 		Key     string
 		Op      selection.Operator
 		Vals    sets.String
-		WantErr field.ErrorList
+		WantErr string
 	}{
 		{
 			Key:     "x1",
 			Op:      selection.In,
-			WantErr: field.ErrorList{
+			WantErr: "values: Invalid value: []string{}: for 'in', 'notin' operators, values set can't be empty",
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values",
-					BadValue: []string{},
-				},
-			},
 		},
 		{
 			Key:     "x2",
 			Op:      selection.NotIn,
 			Vals:    sets.NewString(),
-			WantErr: field.ErrorList{
+			WantErr: "values: Invalid value: []string{}: for 'in', 'notin' operators, values set can't be empty",
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values",
-					BadValue: []string{},
-				},
-			},
 		},
 		{
 			Key:  "x3",
@@ -301,13 +302,7 @@ func TestRequirementConstructor(t *testing.T) {
 			Key:     "x5",
 			Op:      selection.Equals,
 			Vals:    sets.NewString("foo", "bar"),
-			WantErr: field.ErrorList{
+			WantErr: "values: Invalid value: []string{\"bar\", \"foo\"}: exact-match compatibility requires one single value",
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values",
-					BadValue: []string{"bar", "foo"},
-				},
-			},
 		},
 		{
 			Key: "x6",
@@ -321,13 +316,7 @@ func TestRequirementConstructor(t *testing.T) {
 			Key:     "x8",
 			Op:      selection.Exists,
 			Vals:    sets.NewString("foo"),
-			WantErr: field.ErrorList{
+			WantErr: `values: Invalid value: []string{"foo"}: values set must be empty for exists and does not exist`,
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values",
-					BadValue: []string{"foo"},
-				},
-			},
 		},
 		{
 			Key:  "x9",
@@ -352,37 +341,19 @@ func TestRequirementConstructor(t *testing.T) {
 		{
 			Key:     "x13",
 			Op:      selection.GreaterThan,
-			WantErr: field.ErrorList{
+			WantErr: "values: Invalid value: []string{}: for 'Gt', 'Lt' operators, exactly one value is required",
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values",
-					BadValue: []string{},
-				},
-			},
 		},
 		{
 			Key:     "x14",
 			Op:      selection.GreaterThan,
 			Vals:    sets.NewString("bar"),
-			WantErr: field.ErrorList{
+			WantErr: `values[0]: Invalid value: "bar": for 'Gt', 'Lt' operators, the value must be an integer`,
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values[0]",
-					BadValue: "bar",
-				},
-			},
 		},
 		{
 			Key:     "x15",
 			Op:      selection.LessThan,
 			Vals:    sets.NewString("bar"),
-			WantErr: field.ErrorList{
+			WantErr: `values[0]: Invalid value: "bar": for 'Gt', 'Lt' operators, the value must be an integer`,
-				&field.Error{
-					Type:     field.ErrorTypeInvalid,
-					Field:    "values[0]",
-					BadValue: "bar",
-				},
-			},
 		},
 		{
 			Key: strings.Repeat("a", 254), //breaks DNS rule that len(key) <= 253
@@ -401,19 +372,27 @@ func TestRequirementConstructor(t *testing.T) {
 		{
 			Key:     "x18",
 			Op:      "unsupportedOp",
-			WantErr: field.ErrorList{
+			WantErr: `operator: Unsupported value: "unsupportedOp": supported values: "in", "notin", "=", "==", "!=", "~", "!~", "gt", "lt", "=>", "exists", "!"`,
-				&field.Error{
-					Type:     field.ErrorTypeNotSupported,
-					Field:    "operator",
-					BadValue: selection.Operator("unsupportedOp"),
-				},
-			},
 		},
 	}
 	for _, rc := range requirementConstructorTests {
 		_, err := NewRequirement(rc.Key, rc.Op, rc.Vals.List())
-		if diff := cmp.Diff(rc.WantErr.ToAggregate(), err, ignoreDetail); diff != "" {
+		if rc.WantErr != "" {
-			t.Errorf("NewRequirement test %v returned unexpected error (-want,+got):\n%s", rc.Key, diff)
+			assert.NotNil(t, err)
+			if err != nil {
+				assert.Equal(t, rc.WantErr, err.Error())
+			}
+		} else {
+			assert.Nil(t, err)
+		}
+		_, err = NewIndirectRequirement(rc.Key, []string{"herb", "job", "nice", "reading"}, &rc.Op, rc.Vals.List())
+		if rc.WantErr != "" {
+			assert.NotNil(t, err)
+			if err != nil {
+				assert.Equal(t, rc.WantErr, err.Error())
+			}
+		} else {
+			assert.Nil(t, err)
 		}
 	}
 }

pkg/stores/sqlpartition/selection/operator.go

@@ -38,4 +38,5 @@ const (
 	Exists           Operator = "exists"
 	GreaterThan      Operator = "gt"
 	LessThan         Operator = "lt"
+	IndirectSelector Operator = "=>"
 )