with query

pkg/registry/apis/folders/register.go

@@ -129,6 +129,23 @@ func (b *FolderAPIBuilder) InstallSchema(scheme *runtime.Scheme) error {
 		Version: runtime.APIVersionInternal,
 	})
 
+	// Allow searching by owner reference
+	gvk := gv.WithKind("Folder")
+	err := scheme.AddFieldLabelConversionFunc(
+		gvk,
+		func(label, value string) (string, string, error) {
+			if label == "metadata.name" || label == "metadata.namespace" {
+				return label, value, nil
+			}
+			if label == "search.ownerReference" { // TODO: this should become more general
+				return label, value, nil
+			}
+			return "", "", fmt.Errorf("field label not supported for %s: %s", gvk, label)
+		})
+	if err != nil {
+		return err
+	}
+
 	// If multiple versions exist, then register conversions from zz_generated.conversion.go
 	// if err := playlist.RegisterConversions(scheme); err != nil {
 	//   return err

pkg/storage/unified/apistore/util.go

@@ -11,6 +11,7 @@ import (
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apiserver/pkg/storage"
 
@@ -120,6 +121,22 @@ func toListRequest(k *resourcepb.ResourceKey, opts storage.ListOptions) (*resour
 	if opts.Predicate.Field != nil && !opts.Predicate.Field.Empty() {
 		requirements := opts.Predicate.Field.Requirements()
 		for _, r := range requirements {
+			// NOTE: requires: scheme.AddFieldLabelConversionFunc(
+			if r.Field == "search.ownerReference" {
+				if len(requirements) > 1 {
+					return nil, predicate, apierrors.NewBadRequest("search.ownerReference only supports one requirement")
+				}
+				req.Options.Fields = []*resourcepb.Requirement{{
+					Key:      r.Field,
+					Operator: string(r.Operator),
+					Values:   []string{r.Value},
+				}}
+
+				// with only one requirement, we do not need to transform the predicate to exclude this pseudo field
+				predicate.Field = fields.Everything()
+				break
+			}
+
 			requirement := &resourcepb.Requirement{Key: r.Field, Operator: string(r.Operator)}
 			if r.Value != "" {
 				requirement.Values = append(requirement.Values, r.Value)

pkg/storage/unified/resource/fieldSelection.go

@@ -0,0 +1,81 @@
+package resource
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+)
+
+// Some list queries can be calculated with simple reads or search index
+func (s *server) tryFieldSelector(ctx context.Context, req *resourcepb.ListRequest) *resourcepb.ListResponse {
+	if req.Source != resourcepb.ListRequest_STORE || req.Options.Key.Namespace == "" {
+		return nil
+	}
+
+	var names []string
+	for _, v := range req.Options.Fields {
+		if v.Key == "metadata.name" && v.Operator == `=` {
+			names = v.Values
+		}
+
+		// Search by owner reference
+		if v.Key == "search.ownerReference" {
+			if len(req.Options.Fields) > 1 {
+				return &resourcepb.ListResponse{
+					Error: NewBadRequestError("multiple fields found"),
+				}
+			}
+
+			results, err := s.Search(ctx, &resourcepb.ResourceSearchRequest{
+				Fields: []string{}, // no extra fields
+				Options: &resourcepb.ListOptions{
+					Key: req.Options.Key,
+					Fields: []*resourcepb.Requirement{{
+						Key:      SEARCH_FIELD_OWNER_REFERENCES,
+						Operator: v.Operator,
+						Values:   v.Values,
+					}},
+				},
+			})
+			if err != nil {
+				return &resourcepb.ListResponse{
+					Error: AsErrorResult(err),
+				}
+			}
+			if len(results.Results.Rows) < 1 { // nothing found
+				return &resourcepb.ListResponse{
+					ResourceVersion: 1, // TODO, search result should include when it was indexed
+				}
+			}
+			for _, res := range results.Results.Rows {
+				names = append(names, res.Key.Name)
+			}
+		}
+	}
+
+	// The required names
+	if len(names) > 0 {
+		read := &resourcepb.ReadRequest{
+			Key:             req.Options.Key,
+			ResourceVersion: req.ResourceVersion,
+		}
+		rsp := &resourcepb.ListResponse{
+			ResourceVersion: 1, // TODO, search result should include when it was indexed
+		}
+		for _, name := range names {
+			read.Key.Name = name
+			found, err := s.Read(ctx, read)
+			if err != nil {
+				return &resourcepb.ListResponse{Error: AsErrorResult(err)}
+			}
+			if len(found.Value) > 0 {
+				rsp.Items = append(rsp.Items, &resourcepb.ResourceWrapper{
+					Value:           found.Value,
+					ResourceVersion: found.ResourceVersion,
+				})
+			}
+		}
+		return rsp
+	}
+	return nil
+}

pkg/storage/unified/resource/server.go

@@ -22,7 +22,6 @@ import (
 
 	claims "github.com/grafana/authlib/types"
 	"github.com/grafana/dskit/backoff"
-
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
 	"github.com/grafana/grafana/pkg/apimachinery/validation"
 	"github.com/grafana/grafana/pkg/infra/log"
@@ -1038,7 +1037,7 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 	}
 
 	// Fast path for getting single value in a list
-	if rsp := s.tryFastPathList(ctx, req); rsp != nil {
+	if rsp := s.tryFieldSelector(ctx, req); rsp != nil {
 		return rsp, nil
 	}
 
@@ -1136,40 +1135,6 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 	return rsp, err
 }
 
-// Some list queries can be calculated with simple reads
-func (s *server) tryFastPathList(ctx context.Context, req *resourcepb.ListRequest) *resourcepb.ListResponse {
-	if req.Source != resourcepb.ListRequest_STORE || req.Options.Key.Namespace == "" {
-		return nil
-	}
-
-	for _, v := range req.Options.Fields {
-		if v.Key == "metadata.name" && v.Operator == `=` {
-			if len(v.Values) == 1 {
-				read := &resourcepb.ReadRequest{
-					Key:             req.Options.Key,
-					ResourceVersion: req.ResourceVersion,
-				}
-				read.Key.Name = v.Values[0]
-				found, err := s.Read(ctx, read)
-				if err != nil {
-					return &resourcepb.ListResponse{Error: AsErrorResult(err)}
-				}
-
-				// Return a value when it exists
-				rsp := &resourcepb.ListResponse{}
-				if len(found.Value) > 0 {
-					rsp.Items = []*resourcepb.ResourceWrapper{{
-						Value:           found.Value,
-						ResourceVersion: found.ResourceVersion,
-					}}
-				}
-				return rsp
-			}
-		}
-	}
-	return nil
-}
-
 // isTrashItemAuthorized checks if the user has access to the trash item.
 func (s *server) isTrashItemAuthorized(ctx context.Context, iter ListIterator, trashChecker claims.ItemChecker) bool {
 	user, ok := claims.AuthInfoFrom(ctx)

pkg/storage/unified/search/bleve.go

@@ -1559,17 +1559,20 @@ var termFields = []string{
 // Convert a "requirement" into a bleve query
 func requirementQuery(req *resourcepb.Requirement, prefix string) (query.Query, *resourcepb.ErrorResult) {
 	switch selection.Operator(req.Operator) {
-	case selection.Equals, selection.DoubleEquals:
+	case selection.Equals:
 		if len(req.Values) == 0 {
 			return query.NewMatchAllQuery(), nil
 		}
 
 		// FIXME: special case for login and email to use term query only because those fields are using keyword analyzer
 		// This should be fixed by using the info from the schema
-		if (req.Key == "login" || req.Key == "email") && len(req.Values) == 1 {
-			tq := bleve.NewTermQuery(req.Values[0])
-			tq.SetField(prefix + req.Key)
-			return tq, nil
+		if len(req.Values) == 1 {
+			switch req.Key {
+			case "login", "email", resource.SEARCH_FIELD_OWNER_REFERENCES:
+				tq := bleve.NewTermQuery(req.Values[0])
+				tq.SetField(prefix + req.Key)
+				return tq, nil
+			}
 		}
 
 		if len(req.Values) == 1 {
@@ -1585,11 +1588,6 @@ func requirementQuery(req *resourcepb.Requirement, prefix string) (query.Query,
 
 		return query.NewConjunctionQuery(conjuncts), nil
 
-	case selection.NotEquals:
-	case selection.DoesNotExist:
-	case selection.GreaterThan:
-	case selection.LessThan:
-	case selection.Exists:
 	case selection.In:
 		if len(req.Values) == 0 {
 			return query.NewMatchAllQuery(), nil
@@ -1622,6 +1620,14 @@ func requirementQuery(req *resourcepb.Requirement, prefix string) (query.Query,
 		boolQuery.AddMust(notEmptyQuery)
 
 		return boolQuery, nil
+
+	// will fall through to the BadRequestError
+	case selection.DoubleEquals:
+	case selection.NotEquals:
+	case selection.DoesNotExist:
+	case selection.GreaterThan:
+	case selection.LessThan:
+	case selection.Exists:
 	}
 	return nil, resource.NewBadRequestError(
 		fmt.Sprintf("unsupported query operation (%s %s %v)", req.Key, req.Operator, req.Values),

pkg/tests/apis/folder/folders_test.go

@@ -2175,3 +2175,79 @@ func TestIntegrationProvisionedFolderPropagatesLabelsAndAnnotations(t *testing.T
 	require.Equal(t, expectedLabels, accessor.GetLabels())
 	require.Equal(t, expectedAnnotations, accessor.GetAnnotations())
 }
+
+// Test finding folders with an owner
+func TestIntegrationFolderWithOwner(t *testing.T) {
+	helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
+		DisableAnonymous:     true,
+		AppModeProduction:    true,
+		APIServerStorageType: "unified",
+		UnifiedStorageConfig: map[string]setting.UnifiedStorageConfig{
+			folders.RESOURCEGROUP: {
+				DualWriterMode: grafanarest.Mode5,
+			},
+			"dashboards.dashboard.grafana.app": {
+				DualWriterMode: grafanarest.Mode5,
+			},
+		},
+		EnableFeatureToggles: []string{
+			featuremgmt.FlagUnifiedStorageSearch,
+		},
+	})
+	client := helper.GetResourceClient(apis.ResourceClientArgs{
+		User: helper.Org1.Admin,
+		GVR:  gvr,
+	})
+
+	// Without owner
+	folder := &unstructured.Unstructured{
+		Object: map[string]any{
+			"spec": map[string]any{
+				"title": "Folder without owner",
+			},
+		},
+	}
+	folder.SetName("folderA")
+	out, err := client.Resource.Create(context.Background(), folder, metav1.CreateOptions{})
+	require.NoError(t, err)
+	require.Equal(t, folder.GetName(), out.GetName())
+
+	// with owner
+	folder = &unstructured.Unstructured{
+		Object: map[string]any{
+			"spec": map[string]any{
+				"title": "Folder with owner",
+			},
+		},
+	}
+	folder.SetName("folderB")
+	folder.SetOwnerReferences([]metav1.OwnerReference{{
+		APIVersion: "iam.grafana.app/v0alpha1",
+		Kind:       "Team",
+		Name:       "engineering",
+		UID:        "123456", // required by k8s
+	}})
+	out, err = client.Resource.Create(context.Background(), folder, metav1.CreateOptions{})
+	require.NoError(t, err)
+	require.Equal(t, folder.GetName(), out.GetName())
+
+	// Get everything
+	results, err := client.Resource.List(context.Background(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Equal(t, []string{"folderA", "folderB"}, getNames(results.Items))
+
+	// Find results with a specific owner
+	results, err = client.Resource.List(context.Background(), metav1.ListOptions{
+		FieldSelector: "search.ownerReference=iam.grafana.app/Team/engineering",
+	})
+	require.NoError(t, err)
+	require.Equal(t, []string{"folderB"}, getNames(results.Items))
+}
+
+func getNames(items []unstructured.Unstructured) []string {
+	names := make([]string, 0, len(items))
+	for _, item := range items {
+		names = append(names, item.GetName())
+	}
+	return names
+}