Add parallel segment delete for Greenplum (#1300)

* initial * fix linter * add docs
wal-g · Jun 21, 2022 · 3b3ec01 · 3b3ec01
1 parent df3929f
commit 3b3ec01
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 31 deletions.
diff --git a/docs/Greenplum.md b/docs/Greenplum.md
@@ -125,3 +125,6 @@ WAL-G can also do in-place backup restoration without the restore config. It mig
 ```bash
 wal-g backup-fetch LATEST --in-place --config=/path/to/config.yaml
 ```
+
+#### Delete concurrency
+During the delete execution, WAL-G can process segments in parallel mode. To control, how many segments will be processed simultaneously, use the `WALG_GP_DELETE_CONCURRENCY` setting. The default value is `1`. 
diff --git a/internal/config.go b/internal/config.go
@@ -120,6 +120,7 @@ const (
 	GPSegmentsPollRetries  = "WALG_GP_SEG_POLL_RETRIES"
 	GPSegmentsUpdInterval  = "WALG_GP_SEG_UPD_INTERVAL"
 	GPSegmentStatesDir     = "WALG_GP_SEG_STATES_DIR"
+	GPDeleteConcurrency    = "WALG_GP_DELETE_CONCURRENCY"
 
 	GoMaxProcs = "GOMAXPROCS"
 
@@ -207,6 +208,7 @@ var (
 		GPSegmentsUpdInterval:  "10s",
 		GPSegmentsPollRetries:  "5",
 		GPSegmentStatesDir:     "/tmp",
+		GPDeleteConcurrency:    "1",
 	}
 
 	AllowedSettings map[string]bool
@@ -407,6 +409,7 @@ var (
 		GPSegmentsPollInterval: true,
 		GPSegmentsUpdInterval:  true,
 		GPSegmentStatesDir:     true,
+		GPDeleteConcurrency:    true,
 	}
 
 	RequiredSettings       = make(map[string]bool)

diff --git a/internal/databases/greenplum/delete_handler.go b/internal/databases/greenplum/delete_handler.go
@@ -1,11 +1,15 @@
 package greenplum
 
 import (
+	"context"
+	"fmt"
 	"os"
 	"path"
 	"strconv"
 	"strings"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal"
 	"github.com/wal-g/wal-g/internal/databases/postgres"
@@ -113,49 +117,70 @@ func (h *DeleteHandler) DeleteBeforeTarget(target internal.BackupObject, confirm
 	}
 
 	tracelog.InfoLogger.Println("Deleting the segments backups...")
+	errorGroup, _ := errgroup.WithContext(context.Background())
+
+	deleteConcurrency, err := internal.GetMaxConcurrency(internal.GPDeleteConcurrency)
+	if err != nil {
+		tracelog.WarningLogger.Printf("config error: %v", err)
+	}
+
+	deleteSem := make(chan struct{}, deleteConcurrency)
+
 	// clean the segments
 	for i := range sentinel.Segments {
-		meta := &sentinel.Segments[i]
-		tracelog.InfoLogger.Printf("Processing segment %d (backupId=%s)\n", meta.ContentID, meta.BackupID)
+		meta := sentinel.Segments[i]
+		errorGroup.Go(func() error {
+			deleteSem <- struct{}{}
+			deleteErr := h.runDeleteOnSegment(backup, meta, confirmed)
+			<-deleteSem
+			return deleteErr
+		})
+	}
 
-		segFolder := h.Folder.GetSubFolder(FormatSegmentStoragePrefix(meta.ContentID))
-		permanentBackups, permanentWals := postgres.GetPermanentBackupsAndWals(segFolder)
+	err = errorGroup.Wait()
+	if err != nil {
+		return fmt.Errorf("failed to delete the segments backups: %w", err)
+	}
 
-		segDeleteHandler, err := postgres.NewDeleteHandler(segFolder, permanentBackups, permanentWals, false)
-		if err != nil {
-			return err
-		}
+	tracelog.InfoLogger.Printf("Finished deleting the segments backups")
 
-		pgBackup, err := backup.GetSegmentBackup(meta.BackupID, meta.ContentID)
-		if err != nil {
-			return err
-		}
+	return h.DeleteHandler.DeleteBeforeTarget(target, confirmed)
+}
 
-		segTarget, err := segDeleteHandler.FindTargetByName(pgBackup.Name)
-		if err != nil {
-			return err
-		}
+func (h *DeleteHandler) runDeleteOnSegment(backup Backup, meta SegmentMetadata, confirmed bool) error {
+	tracelog.InfoLogger.Printf("Processing segment %d (backupId=%s)\n", meta.ContentID, meta.BackupID)
 
-		tracelog.InfoLogger.Printf("Running delete before target %s on segment %d\n",
-			segTarget.GetBackupName(), meta.ContentID)
+	segFolder := h.Folder.GetSubFolder(FormatSegmentStoragePrefix(meta.ContentID))
+	permanentBackups, permanentWals := postgres.GetPermanentBackupsAndWals(segFolder)
 
-		filterFunc := func(object storage.Object) bool {
-			return !strings.HasSuffix(object.GetName(), postgres.AoSegSuffix)
-		}
+	segDeleteHandler, err := postgres.NewDeleteHandler(segFolder, permanentBackups, permanentWals, false)
+	if err != nil {
+		return err
+	}
 
-		err = segDeleteHandler.DeleteBeforeTargetWhere(segTarget, confirmed, filterFunc)
-		if err != nil {
-			return err
-		}
+	pgBackup, err := backup.GetSegmentBackup(meta.BackupID, meta.ContentID)
+	if err != nil {
+		return err
+	}
 
-		err = cleanupAOSegments(segFolder, confirmed)
-		if err != nil {
-			return err
-		}
+	segTarget, err := segDeleteHandler.FindTargetByName(pgBackup.Name)
+	if err != nil {
+		return err
 	}
-	tracelog.InfoLogger.Printf("Finished deleting the segments backups")
 
-	return h.DeleteHandler.DeleteBeforeTarget(target, confirmed)
+	tracelog.InfoLogger.Printf("Running delete before target %s on segment %d\n",
+		segTarget.GetBackupName(), meta.ContentID)
+
+	filterFunc := func(object storage.Object) bool {
+		return !strings.HasSuffix(object.GetName(), postgres.AoSegSuffix)
+	}
+
+	err = segDeleteHandler.DeleteBeforeTargetWhere(segTarget, confirmed, filterFunc)
+	if err != nil {
+		return err
+	}
+
+	return cleanupAOSegments(segFolder, confirmed)
 }
 
 func cleanupAOSegments(segFolder storage.Folder, confirmed bool) error {