Folders: Reduce DB queries when counting and deleting resources under folders (#81153)

* Add folder store method for fetching all folder descendants

* Modify GetDescendantCounts() to fetch folder descendants at once

* Reduce DB calls when counting library panels under dashboard

* Reduce DB calls when counting dashboards under folder

* Reduce DB calls during folder delete

* Modify folder registry to count/delete entities under multiple folders

* Reduce DB calls when counting

* Reduce DB calls when deleting
This commit is contained in:
Sofia Papagiannaki
2024-01-30 18:26:34 +02:00
committed by GitHub
parent 0139ac205d
commit 89d3b55bec
24 changed files with 335 additions and 411 deletions
+72 -3
View File
@@ -83,11 +83,20 @@ func (ss *sqlStore) Create(ctx context.Context, cmd folder.CreateFolderCommand)
return foldr.WithURL(), err
}
func (ss *sqlStore) Delete(ctx context.Context, uid string, orgID int64) error {
func (ss *sqlStore) Delete(ctx context.Context, UIDs []string, orgID int64) error {
if len(UIDs) == 0 {
return nil
}
return ss.db.WithDbSession(ctx, func(sess *db.Session) error {
_, err := sess.Exec("DELETE FROM folder WHERE uid=? AND org_id=?", uid, orgID)
s := fmt.Sprintf("DELETE FROM folder WHERE org_id=? AND uid IN (%s)", strings.Repeat("?, ", len(UIDs)-1)+"?")
sqlArgs := make([]any, 0, len(UIDs)+2)
sqlArgs = append(sqlArgs, s, orgID)
for _, uid := range UIDs {
sqlArgs = append(sqlArgs, uid)
}
_, err := sess.Exec(sqlArgs...)
if err != nil {
return folder.ErrDatabaseError.Errorf("failed to delete folder: %w", err)
return folder.ErrDatabaseError.Errorf("failed to delete folders: %w", err)
}
return nil
})
@@ -331,6 +340,7 @@ func (ss *sqlStore) getParentsMySQL(ctx context.Context, q folder.GetParentsQuer
return util.Reverse(folders), err
}
// TODO use a single query to get the height of a folder
func (ss *sqlStore) GetHeight(ctx context.Context, foldrUID string, orgID int64, parentUID *string) (int, error) {
height := -1
queue := []string{foldrUID}
@@ -460,6 +470,65 @@ func (ss *sqlStore) GetFolders(ctx context.Context, q getFoldersQuery) ([]*folde
return folders, nil
}
func (ss *sqlStore) GetDescendants(ctx context.Context, orgID int64, ancestor_uid string) ([]*folder.Folder, error) {
var folders []*folder.Folder
recursiveQueriesAreSupported, err := ss.db.RecursiveQueriesAreSupported()
if err != nil {
return nil, err
}
switch recursiveQueriesAreSupported {
case true:
recQuery := `
WITH RECURSIVE RecQry AS (
SELECT * FROM folder WHERE parent_uid = ? AND org_id = ?
UNION ALL SELECT f.* FROM folder f INNER JOIN RecQry r ON f.parent_uid = r.uid and f.org_id = r.org_id
)
SELECT * FROM RecQry;
`
if err := ss.db.WithDbSession(ctx, func(sess *db.Session) error {
err := sess.SQL(recQuery, ancestor_uid, orgID).Find(&folders)
if err != nil {
return folder.ErrDatabaseError.Errorf("failed to get folder descendants: %w", err)
}
return nil
}); err != nil {
return nil, err
}
default:
// this is suboptimal because results is full table scan on f0
// but it's the best we can do without recursive CTE
if err := ss.db.WithDbSession(ctx, func(sess *db.Session) error {
s := strings.Builder{}
args := make([]any, 0, 1+folder.MaxNestedFolderDepth)
args = append(args, orgID)
s.WriteString(`SELECT f0.id, f0.org_id, f0.uid, f0.parent_uid, f0.title, f0.description, f0.created, f0.updated`)
s.WriteString(` FROM folder f0`)
s.WriteString(getFullpathJoinsSQL())
s.WriteString(` WHERE f0.org_id=?`)
s.WriteString(` AND (`)
for i := 1; i <= folder.MaxNestedFolderDepth; i++ {
if i > 1 {
s.WriteString(` OR `)
}
s.WriteString(fmt.Sprintf(`f%d.uid=?`, i))
args = append(args, ancestor_uid)
}
s.WriteString(`)`)
return sess.SQL(s.String(), args...).Find(&folders)
}); err != nil {
return nil, err
}
}
// Add URLs
for i, f := range folders {
folders[i] = f.WithURL()
}
return folders, nil
}
func getFullpathSQL(dialect migrator.Dialect) string {
concatCols := make([]string, 0, folder.MaxNestedFolderDepth)
concatCols = append(concatCols, "COALESCE(REPLACE(f0.title, '/', '\\/'), '')")