Generate sql with indirect queries.

pkg/sqlcache/informer/listoption_indexer.go

@@ -7,7 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
-	"sort"
 	"strconv"
 	"strings"
 
@@ -253,6 +252,8 @@ func (l *ListOptionIndexer) ListByOptions(ctx context.Context, lo *sqltypes.List
 	if err != nil {
 		return nil, 0, "", err
 	}
+	logrus.Debugf("ListOptionIndexer prepared statement: %v", queryInfo.query)
+	logrus.Debugf("Params: %v", queryInfo.params)
 	return l.executeQuery(ctx, queryInfo)
 }
 
@@ -267,210 +268,6 @@ type QueryInfo struct {
 	offset      int
 }
 
-func (l *ListOptionIndexer) constructQuery(lo *sqltypes.ListOptions, partitions []partition.Partition, namespace string, dbName string) (*QueryInfo, error) {
-	ensureSortLabelsAreSelected(lo)
-	queryInfo := &QueryInfo{}
-	queryUsesLabels := hasLabelFilter(lo.Filters)
-	joinTableIndexByLabelName := make(map[string]int)
-
-	// First, what kind of filtering will we be doing?
-	// 1- Intro: SELECT and JOIN clauses
-	// There's a 1:1 correspondence between a base table and its _Fields table
-	// but it's possible that a key has no associated labels, so if we're doing a
-	// non-existence test on labels we need to do a LEFT OUTER JOIN
-	distinctModifier := ""
-	if queryUsesLabels {
-		distinctModifier = " DISTINCT"
-	}
-	query := fmt.Sprintf(`SELECT%s o.object, o.objectnonce, o.dekid FROM "%s" o`, distinctModifier, dbName)
-	query += "\n  "
-	query += fmt.Sprintf(`JOIN "%s_fields" f ON o.key = f.key`, dbName)
-	if queryUsesLabels {
-		for i, orFilter := range lo.Filters {
-			for j, filter := range orFilter.Filters {
-				if isLabelFilter(&filter) {
-					labelName := filter.Field[2]
-					_, ok := joinTableIndexByLabelName[labelName]
-					if !ok {
-						// Make the lt index 1-based for readability
-						jtIndex := i + j + 1
-						joinTableIndexByLabelName[labelName] = jtIndex
-						query += "\n  "
-						query += fmt.Sprintf(`LEFT OUTER JOIN "%s_labels" lt%d ON o.key = lt%d.key`, dbName, jtIndex, jtIndex)
-					}
-				}
-			}
-		}
-	}
-	params := []any{}
-
-	// 2- Filtering: WHERE clauses (from lo.Filters)
-	whereClauses := []string{}
-	for _, orFilters := range lo.Filters {
-		orClause, orParams, err := l.buildORClauseFromFilters(orFilters, dbName, joinTableIndexByLabelName)
-		if err != nil {
-			return queryInfo, err
-		}
-		if orClause == "" {
-			continue
-		}
-		whereClauses = append(whereClauses, orClause)
-		params = append(params, orParams...)
-	}
-
-	// WHERE clauses (from namespace)
-	if namespace != "" && namespace != "*" {
-		whereClauses = append(whereClauses, fmt.Sprintf(`f."metadata.namespace" = ?`))
-		params = append(params, namespace)
-	}
-
-	// WHERE clauses (from partitions and their corresponding parameters)
-	partitionClauses := []string{}
-	for _, thisPartition := range partitions {
-		if thisPartition.Passthrough {
-			// nothing to do, no extra filtering to apply by definition
-		} else {
-			singlePartitionClauses := []string{}
-
-			// filter by namespace
-			if thisPartition.Namespace != "" && thisPartition.Namespace != "*" {
-				singlePartitionClauses = append(singlePartitionClauses, fmt.Sprintf(`f."metadata.namespace" = ?`))
-				params = append(params, thisPartition.Namespace)
-			}
-
-			// optionally filter by names
-			if !thisPartition.All {
-				names := thisPartition.Names
-
-				if len(names) == 0 {
-					// degenerate case, there will be no results
-					singlePartitionClauses = append(singlePartitionClauses, "FALSE")
-				} else {
-					singlePartitionClauses = append(singlePartitionClauses, fmt.Sprintf(`f."metadata.name" IN (?%s)`, strings.Repeat(", ?", len(thisPartition.Names)-1)))
-					// sort for reproducibility
-					sortedNames := thisPartition.Names.UnsortedList()
-					sort.Strings(sortedNames)
-					for _, name := range sortedNames {
-						params = append(params, name)
-					}
-				}
-			}
-
-			if len(singlePartitionClauses) > 0 {
-				partitionClauses = append(partitionClauses, strings.Join(singlePartitionClauses, " AND "))
-			}
-		}
-	}
-	if len(partitions) == 0 {
-		// degenerate case, there will be no results
-		whereClauses = append(whereClauses, "FALSE")
-	}
-	if len(partitionClauses) == 1 {
-		whereClauses = append(whereClauses, partitionClauses[0])
-	}
-	if len(partitionClauses) > 1 {
-		whereClauses = append(whereClauses, "(\n      ("+strings.Join(partitionClauses, ") OR\n      (")+")\n)")
-	}
-
-	if len(whereClauses) > 0 {
-		query += "\n  WHERE\n    "
-		for index, clause := range whereClauses {
-			query += fmt.Sprintf("(%s)", clause)
-			if index == len(whereClauses)-1 {
-				break
-			}
-			query += " AND\n    "
-		}
-	}
-
-	// before proceeding, save a copy of the query and params without LIMIT/OFFSET/ORDER info
-	// for COUNTing all results later
-	countQuery := fmt.Sprintf("SELECT COUNT(*) FROM (%s)", query)
-	countParams := params[:]
-
-	// 3- Sorting: ORDER BY clauses (from lo.Sort)
-	if len(lo.SortList.SortDirectives) > 0 {
-		orderByClauses := []string{}
-		for _, sortDirective := range lo.SortList.SortDirectives {
-			fields := sortDirective.Fields
-			if isLabelsFieldList(fields) {
-				clause, sortParam, err := buildSortLabelsClause(fields[2], joinTableIndexByLabelName, sortDirective.Order == sqltypes.ASC)
-				if err != nil {
-					return nil, err
-				}
-				orderByClauses = append(orderByClauses, clause)
-				params = append(params, sortParam)
-			} else {
-				columnName := toColumnName(fields)
-				if err := l.validateColumn(columnName); err != nil {
-					return queryInfo, err
-				}
-				direction := "ASC"
-				if sortDirective.Order == sqltypes.DESC {
-					direction = "DESC"
-				}
-				orderByClauses = append(orderByClauses, fmt.Sprintf(`f."%s" %s`, columnName, direction))
-			}
-		}
-		query += "\n  ORDER BY "
-		query += strings.Join(orderByClauses, ", ")
-	} else {
-		// make sure one default order is always picked
-		if l.namespaced {
-			query += "\n  ORDER BY f.\"metadata.namespace\" ASC, f.\"metadata.name\" ASC "
-		} else {
-			query += "\n  ORDER BY f.\"metadata.name\" ASC "
-		}
-	}
-
-	// 4- Pagination: LIMIT clause (from lo.Pagination and/or lo.ChunkSize/lo.Resume)
-
-	limitClause := ""
-	// take the smallest limit between lo.Pagination and lo.ChunkSize
-	limit := lo.Pagination.PageSize
-	if limit == 0 || (lo.ChunkSize > 0 && lo.ChunkSize < limit) {
-		limit = lo.ChunkSize
-	}
-	if limit > 0 {
-		limitClause = "\n  LIMIT ?"
-		params = append(params, limit)
-	}
-
-	// OFFSET clause (from lo.Pagination and/or lo.Resume)
-	offsetClause := ""
-	offset := 0
-	if lo.Resume != "" {
-		offsetInt, err := strconv.Atoi(lo.Resume)
-		if err != nil {
-			return queryInfo, err
-		}
-		offset = offsetInt
-	}
-	if lo.Pagination.Page >= 1 {
-		offset += lo.Pagination.PageSize * (lo.Pagination.Page - 1)
-	}
-	if offset > 0 {
-		offsetClause = "\n  OFFSET ?"
-		params = append(params, offset)
-	}
-	if limit > 0 || offset > 0 {
-		query += limitClause
-		query += offsetClause
-		queryInfo.countQuery = countQuery
-		queryInfo.countParams = countParams
-		queryInfo.limit = limit
-		queryInfo.offset = offset
-	}
-	// Otherwise leave these as default values and the executor won't do pagination work
-
-	logrus.Debugf("ListOptionIndexer prepared statement: %v", query)
-	logrus.Debugf("Params: %v", params)
-	queryInfo.query = query
-	queryInfo.params = params
-
-	return queryInfo, nil
-}
-
 func (l *ListOptionIndexer) executeQuery(ctx context.Context, queryInfo *QueryInfo) (result *unstructured.UnstructuredList, total int, token string, err error) {
 	stmt := l.Prepare(queryInfo.query)
 	defer func() {
@@ -528,338 +325,12 @@ func (l *ListOptionIndexer) executeQuery(ctx context.Context, queryInfo *QueryIn
 	return toUnstructuredList(items), total, continueToken, nil
 }
 
-func (l *ListOptionIndexer) validateColumn(column string) error {
-	for _, v := range l.indexedFields {
-		if v == column {
-			return nil
+func extractSubFields(fields string) []string {
+	subfields := make([]string, 0)
+	for _, subField := range subfieldRegex.FindAllString(fields, -1) {
+		subfields = append(subfields, strings.TrimSuffix(subField, "."))
 	}
-	}
-	return fmt.Errorf("column is invalid [%s]: %w", column, ErrInvalidColumn)
-}
-
-// buildORClause creates an SQLite compatible query that ORs conditions built from passed filters
-func (l *ListOptionIndexer) buildORClauseFromFilters(orFilters sqltypes.OrFilter, dbName string, joinTableIndexByLabelName map[string]int) (string, []any, error) {
-	var params []any
-	clauses := make([]string, 0, len(orFilters.Filters))
-	var newParams []any
-	var newClause string
-	var err error
-
-	for _, filter := range orFilters.Filters {
-		if isLabelFilter(&filter) {
-			index, ok := joinTableIndexByLabelName[filter.Field[2]]
-			if !ok {
-				return "", nil, fmt.Errorf("internal error: no index for label name %s", filter.Field[2])
-			}
-			newClause, newParams, err = l.getLabelFilter(index, filter, dbName)
-		} else {
-			newClause, newParams, err = l.getFieldFilter(filter)
-		}
-		if err != nil {
-			return "", nil, err
-		}
-		clauses = append(clauses, newClause)
-		params = append(params, newParams...)
-	}
-	switch len(clauses) {
-	case 0:
-		return "", params, nil
-	case 1:
-		return clauses[0], params, nil
-	}
-	return fmt.Sprintf("(%s)", strings.Join(clauses, ") OR (")), params, nil
-}
-
-func buildSortLabelsClause(labelName string, joinTableIndexByLabelName map[string]int, isAsc bool) (string, string, error) {
-	ltIndex, ok := joinTableIndexByLabelName[labelName]
-	if !ok {
-		return "", "", fmt.Errorf(`internal error: no join-table index given for labelName "%s"`, labelName)
-	}
-	stmt := fmt.Sprintf(`CASE lt%d.label WHEN ? THEN lt%d.value ELSE NULL END`, ltIndex, ltIndex)
-	dir := "ASC"
-	nullsPosition := "LAST"
-	if !isAsc {
-		dir = "DESC"
-		nullsPosition = "FIRST"
-	}
-	return fmt.Sprintf("(%s) %s NULLS %s", stmt, dir, nullsPosition), labelName, nil
-}
-
-// If the user tries to sort on a particular label without mentioning it in a query,
-// it turns out that the sort-directive is ignored. It could be that the sqlite engine
-// is doing some kind of optimization on the `select distinct`, but verifying an otherwise
-// unreferenced label exists solves this problem.
-// And it's better to do this by modifying the ListOptions object.
-// There are no thread-safety issues in doing this because the ListOptions object is
-// created in Store.ListByPartitions, and that ends up calling ListOptionIndexer.ConstructQuery.
-// No other goroutines access this object.
-func ensureSortLabelsAreSelected(lo *sqltypes.ListOptions) {
-	if len(lo.SortList.SortDirectives) == 0 {
-		return
-	}
-	unboundSortLabels := make(map[string]bool)
-	for _, sortDirective := range lo.SortList.SortDirectives {
-		fields := sortDirective.Fields
-		if isLabelsFieldList(fields) {
-			unboundSortLabels[fields[2]] = true
-		}
-	}
-	if len(unboundSortLabels) == 0 {
-		return
-	}
-	// If we have sort directives but no filters, add an exists-filter for each label.
-	if lo.Filters == nil || len(lo.Filters) == 0 {
-		lo.Filters = make([]sqltypes.OrFilter, 1)
-		lo.Filters[0].Filters = make([]sqltypes.Filter, len(unboundSortLabels))
-		i := 0
-		for labelName := range unboundSortLabels {
-			lo.Filters[0].Filters[i] = sqltypes.Filter{
-				Field: []string{"metadata", "labels", labelName},
-				Op:    sqltypes.Exists,
-			}
-			i++
-		}
-		return
-	}
-	// The gotcha is we have to bind the labels for each set of orFilters, so copy them each time
-	for i, orFilters := range lo.Filters {
-		copyUnboundSortLabels := make(map[string]bool, len(unboundSortLabels))
-		for k, v := range unboundSortLabels {
-			copyUnboundSortLabels[k] = v
-		}
-		for _, filter := range orFilters.Filters {
-			if isLabelFilter(&filter) {
-				copyUnboundSortLabels[filter.Field[2]] = false
-			}
-		}
-		// Now for any labels that are still true, add another where clause
-		for labelName, needsBinding := range copyUnboundSortLabels {
-			if needsBinding {
-				// `orFilters` is a copy of lo.Filters[i], so reference the original.
-				lo.Filters[i].Filters = append(lo.Filters[i].Filters, sqltypes.Filter{
-					Field: []string{"metadata", "labels", labelName},
-					Op:    sqltypes.Exists,
-				})
-			}
-		}
-	}
-}
-
-// Possible ops from the k8s parser:
-// KEY = and == (same) VALUE
-// KEY != VALUE
-// KEY exists []  # ,KEY, => this filter
-// KEY ! []  # ,!KEY, => assert KEY doesn't exist
-// KEY in VALUES
-// KEY notin VALUES
-
-func (l *ListOptionIndexer) getFieldFilter(filter sqltypes.Filter) (string, []any, error) {
-	opString := ""
-	escapeString := ""
-	columnName := toColumnName(filter.Field)
-	if err := l.validateColumn(columnName); err != nil {
-		return "", nil, err
-	}
-	switch filter.Op {
-	case sqltypes.Eq:
-		if filter.Partial {
-			opString = "LIKE"
-			escapeString = escapeBackslashDirective
-		} else {
-			opString = "="
-		}
-		clause := fmt.Sprintf(`f."%s" %s ?%s`, columnName, opString, escapeString)
-		return clause, []any{formatMatchTarget(filter)}, nil
-	case sqltypes.NotEq:
-		if filter.Partial {
-			opString = "NOT LIKE"
-			escapeString = escapeBackslashDirective
-		} else {
-			opString = "!="
-		}
-		clause := fmt.Sprintf(`f."%s" %s ?%s`, columnName, opString, escapeString)
-		return clause, []any{formatMatchTarget(filter)}, nil
-
-	case sqltypes.Lt, sqltypes.Gt:
-		sym, target, err := prepareComparisonParameters(filter.Op, filter.Matches[0])
-		if err != nil {
-			return "", nil, err
-		}
-		clause := fmt.Sprintf(`f."%s" %s ?`, columnName, sym)
-		return clause, []any{target}, nil
-
-	case sqltypes.Exists, sqltypes.NotExists:
-		return "", nil, errors.New("NULL and NOT NULL tests aren't supported for non-label queries")
-
-	case sqltypes.In:
-		fallthrough
-	case sqltypes.NotIn:
-		target := "()"
-		if len(filter.Matches) > 0 {
-			target = fmt.Sprintf("(?%s)", strings.Repeat(", ?", len(filter.Matches)-1))
-		}
-		opString = "IN"
-		if filter.Op == sqltypes.NotIn {
-			opString = "NOT IN"
-		}
-		clause := fmt.Sprintf(`f."%s" %s %s`, columnName, opString, target)
-		matches := make([]any, len(filter.Matches))
-		for i, match := range filter.Matches {
-			matches[i] = match
-		}
-		return clause, matches, nil
-	}
-
-	return "", nil, fmt.Errorf("unrecognized operator: %s", opString)
-}
-
-func (l *ListOptionIndexer) getLabelFilter(index int, filter sqltypes.Filter, dbName string) (string, []any, error) {
-	opString := ""
-	escapeString := ""
-	matchFmtToUse := strictMatchFmt
-	labelName := filter.Field[2]
-	switch filter.Op {
-	case sqltypes.Eq:
-		if filter.Partial {
-			opString = "LIKE"
-			escapeString = escapeBackslashDirective
-			matchFmtToUse = matchFmt
-		} else {
-			opString = "="
-		}
-		clause := fmt.Sprintf(`lt%d.label = ? AND lt%d.value %s ?%s`, index, index, opString, escapeString)
-		return clause, []any{labelName, formatMatchTargetWithFormatter(filter.Matches[0], matchFmtToUse)}, nil
-
-	case sqltypes.NotEq:
-		if filter.Partial {
-			opString = "NOT LIKE"
-			escapeString = escapeBackslashDirective
-			matchFmtToUse = matchFmt
-		} else {
-			opString = "!="
-		}
-		subFilter := sqltypes.Filter{
-			Field: filter.Field,
-			Op:    sqltypes.NotExists,
-		}
-		existenceClause, subParams, err := l.getLabelFilter(index, subFilter, dbName)
-		if err != nil {
-			return "", nil, err
-		}
-		clause := fmt.Sprintf(`(%s) OR (lt%d.label = ? AND lt%d.value %s ?%s)`, existenceClause, index, index, opString, escapeString)
-		params := append(subParams, labelName, formatMatchTargetWithFormatter(filter.Matches[0], matchFmtToUse))
-		return clause, params, nil
-
-	case sqltypes.Lt, sqltypes.Gt:
-		sym, target, err := prepareComparisonParameters(filter.Op, filter.Matches[0])
-		if err != nil {
-			return "", nil, err
-		}
-		clause := fmt.Sprintf(`lt%d.label = ? AND lt%d.value %s ?`, index, index, sym)
-		return clause, []any{labelName, target}, nil
-
-	case sqltypes.Exists:
-		clause := fmt.Sprintf(`lt%d.label = ?`, index)
-		return clause, []any{labelName}, nil
-
-	case sqltypes.NotExists:
-		clause := fmt.Sprintf(`o.key NOT IN (SELECT o1.key FROM "%s" o1
-		JOIN "%s_fields" f1 ON o1.key = f1.key
-		LEFT OUTER JOIN "%s_labels" lt%di1 ON o1.key = lt%di1.key
-		WHERE lt%di1.label = ?)`, dbName, dbName, dbName, index, index, index)
-		return clause, []any{labelName}, nil
-
-	case sqltypes.In:
-		target := "(?"
-		if len(filter.Matches) > 0 {
-			target += strings.Repeat(", ?", len(filter.Matches)-1)
-		}
-		target += ")"
-		clause := fmt.Sprintf(`lt%d.label = ? AND lt%d.value IN %s`, index, index, target)
-		matches := make([]any, len(filter.Matches)+1)
-		matches[0] = labelName
-		for i, match := range filter.Matches {
-			matches[i+1] = match
-		}
-		return clause, matches, nil
-
-	case sqltypes.NotIn:
-		target := "(?"
-		if len(filter.Matches) > 0 {
-			target += strings.Repeat(", ?", len(filter.Matches)-1)
-		}
-		target += ")"
-		subFilter := sqltypes.Filter{
-			Field: filter.Field,
-			Op:    sqltypes.NotExists,
-		}
-		existenceClause, subParams, err := l.getLabelFilter(index, subFilter, dbName)
-		if err != nil {
-			return "", nil, err
-		}
-		clause := fmt.Sprintf(`(%s) OR (lt%d.label = ? AND lt%d.value NOT IN %s)`, existenceClause, index, index, target)
-		matches := append(subParams, labelName)
-		for _, match := range filter.Matches {
-			matches = append(matches, match)
-		}
-		return clause, matches, nil
-	}
-	return "", nil, fmt.Errorf("unrecognized operator: %s", opString)
-}
-
-func prepareComparisonParameters(op sqltypes.Op, target string) (string, float64, error) {
-	num, err := strconv.ParseFloat(target, 32)
-	if err != nil {
-		return "", 0, err
-	}
-	switch op {
-	case sqltypes.Lt:
-		return "<", num, nil
-	case sqltypes.Gt:
-		return ">", num, nil
-	}
-	return "", 0, fmt.Errorf("unrecognized operator when expecting '<' or '>': '%s'", op)
-}
-
-func formatMatchTarget(filter sqltypes.Filter) string {
-	format := strictMatchFmt
-	if filter.Partial {
-		format = matchFmt
-	}
-	return formatMatchTargetWithFormatter(filter.Matches[0], format)
-}
-
-func formatMatchTargetWithFormatter(match string, format string) string {
-	// To allow matches on the backslash itself, the character needs to be replaced first.
-	// Otherwise, it will undo the following replacements.
-	match = strings.ReplaceAll(match, `\`, `\\`)
-	match = strings.ReplaceAll(match, `_`, `\_`)
-	match = strings.ReplaceAll(match, `%`, `\%`)
-	return fmt.Sprintf(format, match)
-}
-
-// There are two kinds of string arrays to turn into a string, based on the last value in the array
-// simple: ["a", "b", "conformsToIdentifier"] => "a.b.conformsToIdentifier"
-// complex: ["a", "b", "foo.io/stuff"] => "a.b[foo.io/stuff]"
-
-func smartJoin(s []string) string {
-	if len(s) == 0 {
-		return ""
-	}
-	if len(s) == 1 {
-		return s[0]
-	}
-	lastBit := s[len(s)-1]
-	simpleName := regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
-	if simpleName.MatchString(lastBit) {
-		return strings.Join(s, ".")
-	}
-	return fmt.Sprintf("%s[%s]", strings.Join(s[0:len(s)-1], "."), lastBit)
-}
-
-// toColumnName returns the column name corresponding to a field expressed as string slice
-func toColumnName(s []string) string {
-	return db.Sanitize(smartJoin(s))
+	return subfields
 }
 
 // getField extracts the value of a field expressed as a string path from an unstructured object
@@ -920,31 +391,9 @@ func getField(a any, field string) (any, error) {
 	return obj, nil
 }
 
-func extractSubFields(fields string) []string {
-	subfields := make([]string, 0)
-	for _, subField := range subfieldRegex.FindAllString(fields, -1) {
-		subfields = append(subfields, strings.TrimSuffix(subField, "."))
-	}
-	return subfields
-}
-
-func isLabelFilter(f *sqltypes.Filter) bool {
-	return len(f.Field) >= 2 && f.Field[0] == "metadata" && f.Field[1] == "labels"
-}
-
-func hasLabelFilter(filters []sqltypes.OrFilter) bool {
-	for _, outerFilter := range filters {
-		for _, filter := range outerFilter.Filters {
-			if isLabelFilter(&filter) {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-func isLabelsFieldList(fields []string) bool {
-	return len(fields) == 3 && fields[0] == "metadata" && fields[1] == "labels"
+// toColumnName returns the column name corresponding to a field expressed as string slice
+func toColumnName(s []string) string {
+	return db.Sanitize(smartJoin(s))
 }
 
 // toUnstructuredList turns a slice of unstructured objects into an unstructured.UnstructuredList

pkg/sqlcache/sqltypes/types.go

@@ -25,11 +25,11 @@ const (
 
 // ListOptions represents the query parameters that may be included in a list request.
 type ListOptions struct {
-	ChunkSize  int
-	Resume     string
-	Filters    []OrFilter
-	SortList   SortList
-	Pagination Pagination
+	ChunkSize  int        `json:"chunkSize"`
+	Resume     string     `json:"resume"`
+	Filters    []OrFilter `json:"orFilters"`
+	SortList   SortList   `json:"sortList"`
+	Pagination Pagination `json:"pagination"`
 }
 
 // Filter represents a field to filter by.
@@ -40,12 +40,12 @@ type ListOptions struct {
 //
 // If more than one value is given for the `Match` field, we do an "IN (<values>)" test
 type Filter struct {
-	Field          []string
-	Matches        []string
-	Op             Op
-	Partial        bool
-	IsIndirect     bool
-	IndirectFields []string
+	Field          []string `json:"fields"`
+	Matches        []string `json:"matches"`
+	Op             Op       `json:"op"`
+	Partial        bool     `json:"partial"`
+	IsIndirect     bool     `json:"isIndirect"`
+	IndirectFields []string `json:"indirectFields"`
 }
 
 // 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,20 +59,20 @@ type OrFilter struct {
 // The order is represented by prefixing the sort key by '-', e.g. sort=-metadata.name.
 // e.g. To sort internal clusters first followed by clusters in alpha order: sort=-spec.internal,spec.displayName
 type Sort struct {
-	Fields         []string
-	Order          SortOrder
-	IsIndirect     bool
-	IndirectFields []string
+	Fields         []string  `json:"fields"`
+	Order          SortOrder `json:"order"`
+	IsIndirect     bool      `json:"isIndirect"`
+	IndirectFields []string  `json:"indirectFields"`
 }
 
 type SortList struct {
-	SortDirectives []Sort
+	SortDirectives []Sort `json:"sortDirectives"`
 }
 
 // Pagination represents how to return paginated results.
 type Pagination struct {
-	PageSize int
-	Page     int
+	PageSize int `json:"pageSize"`
+	Page     int `json:"page"`
 }
 
 func NewSortList() *SortList {

pkg/stores/sqlpartition/listprocessor/processor.go

@@ -73,11 +73,14 @@ func k8sRequirementToOrFilter(requirement queryparser.Requirement) (sqltypes.Fil
 	values := requirement.Values()
 	queryFields := splitQuery(requirement.Key())
 	op, usePartialMatch, err := k8sOpToRancherOp(requirement.Operator())
+	isIndirect, indirectFields := requirement.IndirectInfo()
 	return sqltypes.Filter{
 		Field:          queryFields,
 		Matches:        values,
 		Op:             op,
 		Partial:        usePartialMatch,
+		IsIndirect:     isIndirect,
+		IndirectFields: indirectFields,
 	}, err
 }
 

pkg/stores/sqlpartition/listprocessor/processor_test.go

@@ -370,6 +370,33 @@ func TestParseQuery(t *testing.T) {
 			},
 		},
 	})
+	tests = append(tests, testCase{
+		description: "ParseQuery() with an indirect labels filter param should create an indirect labels-specific filter.",
+		req: &types.APIRequest{
+			Request: &http.Request{
+				URL: &url.URL{RawQuery: "filter=metadata.labels[grover.example.com/fish]=>[_v1][Foods][foodCode][country]=japan"},
+			},
+		},
+		expectedLO: sqltypes.ListOptions{
+			ChunkSize: defaultLimit,
+			Filters: []sqltypes.OrFilter{
+				{
+					Filters: []sqltypes.Filter{
+						{
+							Field:          []string{"metadata", "labels", "grover.example.com/fish"},
+							Matches:        []string{"japan"},
+							Op:             sqltypes.Eq,
+							IsIndirect:     true,
+							IndirectFields: []string{"_v1", "Foods", "foodCode", "country"},
+						},
+					},
+				},
+			},
+			Pagination: sqltypes.Pagination{
+				Page: 1,
+			},
+		},
+	})
 	tests = append(tests, testCase{
 		description: "ParseQuery() with multiple filter params, should include multiple or filters.",
 		req: &types.APIRequest{