From c24333fb79f19a691aba260a052b251c3861979e Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Fri, 11 Sep 2026 07:15:47 +0000 Subject: [PATCH] refactor(search): stream bleve descendants Move, delete, restore and purge walk a folder in pages of 20k sorted by id instead of collecting every descendant first: 133 MB -> 19 MB live heap at 100k files, flat over folder size. The walker yields the folder itself, the batch skips it by id. --- services/search/pkg/bleve/batch.go | 149 ++++++------------ services/search/pkg/bleve/descendants_test.go | 94 +++++++++++ services/search/pkg/bleve/index.go | 67 +++----- 3 files changed, 169 insertions(+), 141 deletions(-) create mode 100644 services/search/pkg/bleve/descendants_test.go diff --git a/services/search/pkg/bleve/batch.go b/services/search/pkg/bleve/batch.go index a6966a8a02..62bf4301f0 100644 --- a/services/search/pkg/bleve/batch.go +++ b/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) }) } diff --git a/services/search/pkg/bleve/descendants_test.go b/services/search/pkg/bleve/descendants_test.go new file mode 100644 index 0000000000..8247e273f4 --- /dev/null +++ b/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()) + }), + ) +}) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index e72754b814..8182beeaef 100644 --- a/services/search/pkg/bleve/index.go +++ b/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 }