refactor(search): stream bleve descendants

services/search/pkg/bleve/batch.go

@@ -54,131 +54,84 @@ func (b *Batch) indexResource(id string, r search.Resource) error {
 
 func (b *Batch) Move(id, parentID, location string) error {
 	return b.withSizeLimit(func() error {
-		rootResource, err := searchResourceByID(id, b.index)
-		if err != nil {
-			return err
-		}
-		currentPath := rootResource.Path
 		nextPath := utils.MakeRelativePath(location)
-
-		rootResource.Path = nextPath
-		rootResource.Name = path.Base(nextPath)
-		rootResource.ParentID = parentID
-
-		resources := []*search.Resource{rootResource}
-
-		if rootResource.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
-			descendantResources, err := searchResourcesByPath(rootResource.RootID, currentPath, b.index)
-			if err != nil {
-				return err
-			}
-
-			for _, descendantResource := range descendantResources {
-				descendantResource.Path = strings.Replace(descendantResource.Path, currentPath, nextPath, 1)
-				resources = append(resources, descendantResource)
+		var currentPath string
+		return b.forSelfAndDescendants(id, func(resource *search.Resource) error {
+			if resource.ID == id {
+				currentPath = resource.Path
+				resource.Path = nextPath
+				resource.Name = path.Base(nextPath)
+				resource.ParentID = parentID
+			} else {
+				resource.Path = strings.Replace(resource.Path, currentPath, nextPath, 1)
 			}
-		}
-
-		for _, resource := range resources {
 			resource.Hidden = search.IsHidden(resource.Path)
-
-			if err := b.indexResource(resource.ID, *resource); err != nil {
-				return err
-			}
-			if b.batch.Size() >= b.size {
-				if err := b.Push(); err != nil {
-					return err
-				}
-			}
-		}
-
-		return nil
+			return b.indexResource(resource.ID, *resource)
+		})
 	})
 }
 
 func (b *Batch) Delete(id string) error {
 	return b.withSizeLimit(func() error {
-		affectedResources, err := searchAndUpdateResourcesDeletionState(id, true, b.index)
-		if err != nil {
-			return err
-		}
-
-		for _, resource := range affectedResources {
-			if err := b.indexResource(resource.ID, *resource); err != nil {
-				return err
-			}
-			if b.batch.Size() >= b.size {
-				if err := b.Push(); err != nil {
-					return err
-				}
-			}
-		}
-
-		return nil
+		return b.setDeleted(id, true)
 	})
 }
 
 func (b *Batch) Restore(id string) error {
 	return b.withSizeLimit(func() error {
-		affectedResources, err := searchAndUpdateResourcesDeletionState(id, false, b.index)
-		if err != nil {
-			return err
-		}
-
-		for _, resource := range affectedResources {
-			if err := b.indexResource(resource.ID, *resource); err != nil {
-				return err
-			}
-			if b.batch.Size() >= b.size {
-				if err := b.Push(); err != nil {
-					return err
-				}
-			}
-		}
+		return b.setDeleted(id, false)
+	})
+}
 
-		return nil
+func (b *Batch) setDeleted(id string, deleted bool) error {
+	return b.forSelfAndDescendants(id, func(resource *search.Resource) error {
+		resource.Deleted = deleted
+		return b.indexResource(resource.ID, *resource)
 	})
 }
 
 func (b *Batch) Purge(id string, onlyDeleted bool) error {
 	return b.withSizeLimit(func() error {
-		rootResource, err := searchResourceByID(id, b.index)
-		if err != nil {
-			return err
-		}
-
-		var affectResources []*search.Resource
-		add := func(resource *search.Resource) {
+		return b.forSelfAndDescendants(id, func(resource *search.Resource) error {
 			if onlyDeleted && !resource.Deleted {
-				return
+				return nil
 			}
+			b.batch.Delete(resource.ID)
+			return nil
+		})
+	})
+}
 
-			affectResources = append(affectResources, resource)
-		}
-
-		add(rootResource)
-
-		if rootResource.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
-			descendantResources, err := searchResourcesByPath(rootResource.RootID, rootResource.Path, b.index)
-			if err != nil {
-				return err
-			}
+// fn sees the root first; the root's original path drives the descendant lookup
+func (b *Batch) forSelfAndDescendants(id string, fn func(*search.Resource) error) error {
+	root, err := searchResourceByID(id, b.index)
+	if err != nil {
+		return err
+	}
+	rootID, rootPath := root.RootID, root.Path
+	isContainer := root.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER)
 
-			for _, descendantResource := range descendantResources {
-				add(descendantResource)
-			}
+	apply := func(resource *search.Resource) error {
+		if err := fn(resource); err != nil {
+			return err
 		}
-
-		for _, resource := range affectResources {
-			b.batch.Delete(resource.ID)
-			if b.batch.Size() >= b.size {
-				if err := b.Push(); err != nil {
-					return err
-				}
-			}
+		if b.batch.Size() >= b.size {
+			return b.Push()
 		}
+		return nil
+	}
 
+	if err := apply(root); err != nil {
+		return err
+	}
+	if !isContainer {
 		return nil
+	}
+	return forEachResourceByPath(rootID, rootPath, b.index, func(resource *search.Resource) error {
+		if resource.ID == id {
+			return nil
+		}
+		return apply(resource)
 	})
 }
 

services/search/pkg/bleve/descendants_test.go

@@ -0,0 +1,94 @@
+package bleve
+
+import (
+	"fmt"
+
+	"github.com/blevesearch/bleve/v2"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+func indexResources(idx bleve.Index, resources ...search.Resource) {
+	batch := idx.NewBatch()
+	for _, r := range resources {
+		Expect(batch.Index(r.ID, r)).To(Succeed())
+		if batch.Size() >= 1000 {
+			Expect(idx.Batch(batch)).To(Succeed())
+			batch.Reset()
+		}
+	}
+	Expect(idx.Batch(batch)).To(Succeed())
+}
+
+var _ = Describe("forEachResourceByPath", func() {
+	var idx bleve.Index
+
+	BeforeEach(func() {
+		var err error
+		idx, _, err = NewIndex(GinkgoT().TempDir(), log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		DeferCleanup(func() { Expect(idx.Close()).To(Succeed()) })
+	})
+
+	pageSize := func(n int) {
+		old := descendantPageSize
+		descendantPageSize = n
+		DeferCleanup(func() { descendantPageSize = old })
+	}
+
+	count := func(path string, deleted *bool) uint64 {
+		pq := bleve.NewTermQuery(path)
+		pq.SetField("Path")
+		q := bleve.NewConjunctionQuery(pq)
+		if deleted != nil {
+			dq := bleve.NewBoolFieldQuery(*deleted)
+			dq.SetField("Deleted")
+			q.AddQuery(dq)
+		}
+		req := bleve.NewSearchRequest(q)
+		req.Size = 0
+		res, err := idx.Search(req)
+		Expect(err).ToNot(HaveOccurred())
+		return res.Total
+	}
+	deleted := true
+
+	DescribeTable("walks a folder larger than one page while the batch pushes between pages",
+		func(op func(*Batch) error, check func()) {
+			pageSize(1000)
+			const n, rootID = 6000, "s$op!root"
+			docs := []search.Resource{{ID: "s$op!big", RootID: rootID, Path: "./big", Type: 2}}
+			for i := 0; i < n; i++ {
+				docs = append(docs, search.Resource{ID: fmt.Sprintf("s$op!f%05d", i), RootID: rootID, Path: fmt.Sprintf("./big/f%05d.txt", i), Type: 1})
+			}
+			indexResources(idx, docs...)
+
+			batch, err := NewBatch(idx, 100)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(op(batch)).To(Succeed())
+			Expect(batch.Push()).To(Succeed())
+			check()
+		},
+		Entry("move", func(b *Batch) error { return b.Move("s$op!big", "s$op!root", "./moved") }, func() {
+			Expect(count("./big", nil)).To(BeZero())
+			Expect(count("./moved", nil)).To(Equal(uint64(6001)))
+		}),
+		Entry("delete", func(b *Batch) error { return b.Delete("s$op!big") }, func() {
+			Expect(count("./big", &deleted)).To(Equal(uint64(6001)))
+		}),
+		Entry("purge", func(b *Batch) error {
+			if err := b.Delete("s$op!big"); err != nil {
+				return err
+			}
+			if err := b.Push(); err != nil {
+				return err
+			}
+			return b.Purge("s$op!big", true)
+		}, func() {
+			Expect(count("./big", nil)).To(BeZero())
+		}),
+	)
+})

services/search/pkg/bleve/index.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math"
 	"path/filepath"
 	"reflect"
 	"strings"
@@ -16,7 +15,6 @@ import (
 	"github.com/blevesearch/bleve/v2/analysis/token/lowercase"
 	"github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode"
 	"github.com/blevesearch/bleve/v2/mapping"
-	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve/hierarchy"
@@ -263,55 +261,38 @@ func searchResourceByID(id string, index bleve.Index) (*search.Resource, error)
 	return matchToResource(res.Hits[0]), nil
 }
 
-// searchResourcesByPath returns the descendants of the folder at lookupPath.
-// The folder term matches the folder and everything below it in one term
-// query (see PathAnalyzer); the folder itself is dropped from the result.
-func searchResourcesByPath(rootID string, lookupPath string, index bleve.Index) ([]*search.Resource, error) {
+// every page re-sorts the full match set by id, so a bigger page trades live
+// memory (about 7 MB per 5k hits) for fewer rescans
+var descendantPageSize = 20_000
+
+// forEachResourceByPath streams the folder at lookupPath and its descendants
+// (the folder term matches both, see PathAnalyzer); paged by id so memory is
+// bounded by the page and fn may write to the index between pages
+func forEachResourceByPath(rootID string, lookupPath string, index bleve.Index, fn func(*search.Resource) error) error {
 	rootQuery := bleve.NewTermQuery(rootID)
 	rootQuery.SetField("RootID")
 	pathQuery := bleve.NewTermQuery(lookupPath)
 	pathQuery.SetField("Path")
-	q := bleve.NewConjunctionQuery(rootQuery, pathQuery)
-	bleveReq := bleve.NewSearchRequest(q)
-	bleveReq.Size = math.MaxInt
-	bleveReq.Fields = []string{"*"}
-	res, err := index.Search(bleveReq)
-	if err != nil {
-		return nil, err
-	}
-
-	resources := make([]*search.Resource, 0, res.Hits.Len())
-	for _, match := range res.Hits {
-		resource := matchToResource(match)
-		if resource.Path == lookupPath {
-			continue
-		}
-		resources = append(resources, resource)
-	}
-
-	return resources, nil
-}
-
-func searchAndUpdateResourcesDeletionState(id string, state bool, index bleve.Index) ([]*search.Resource, error) {
-	rootResource, err := searchResourceByID(id, index)
-	if err != nil {
-		return nil, err
-	}
-	rootResource.Deleted = state
 
-	resources := []*search.Resource{rootResource}
+	pageSize := descendantPageSize
+	bleveReq := bleve.NewSearchRequest(bleve.NewConjunctionQuery(rootQuery, pathQuery))
+	bleveReq.Size = pageSize
+	bleveReq.Fields = []string{"*"}
+	bleveReq.SortBy([]string{"_id"})
 
-	if rootResource.Type == uint64(storageProvider.ResourceType_RESOURCE_TYPE_CONTAINER) {
-		descendantResources, err := searchResourcesByPath(rootResource.RootID, rootResource.Path, index)
+	for {
+		res, err := index.Search(bleveReq)
 		if err != nil {
-			return nil, err
+			return err
 		}
-
-		for _, descendantResource := range descendantResources {
-			descendantResource.Deleted = state
-			resources = append(resources, descendantResource)
+		for _, match := range res.Hits {
+			if err := fn(matchToResource(match)); err != nil {
+				return err
+			}
 		}
+		if res.Hits.Len() < pageSize {
+			return nil
+		}
+		bleveReq.SearchAfter = []string{res.Hits[res.Hits.Len()-1].ID}
 	}
-
-	return resources, nil
 }