PBM-1459: PBM ignores index removal during selective PITR restore wit…

…h name remapping (#1063)

Fix indexes cloning logic to support drop op:
During drop operation within oplog, indexes catalog is modified with
cloning-to ns. Therefore it is necessary to have modified index catalog
(with cloned collection) before we start applying the oplog. The approach to
modify index catalog with cloning collection during the snapshot restore
phase, ensures that PITR restore will be correctly applied also on index
catalog.

pbm/oplog/restore.go

@@ -114,8 +114,8 @@ type cloneNS struct {
 
 func (c *cloneNS) SetNSPair(nsPair snapshot.CloneNS) {
 	c.CloneNS = nsPair
-	c.fromDB, c.fromColl, _ = strings.Cut(nsPair.FromNS, ".")
-	c.toDB, c.toColl, _ = strings.Cut(nsPair.ToNS, ".")
+	c.fromDB, c.fromColl = nsPair.SplitFromNS()
+	c.toDB, c.toColl = nsPair.SplitToNS()
 }
 
 // OplogRestore is the oplog applyer
@@ -794,7 +794,7 @@ func (o *OplogRestore) cloneEntry(op *db.Oplog) {
 	}
 
 	cmdName := op.Object[0].Key
-	if cmdName != "create" && cmdName != "drop" {
+	if _, ok := cloningNSSupportedCommands[cmdName]; !ok {
 		return
 	}
 

pbm/restore/logical.go

@@ -273,7 +273,11 @@ func (r *Restore) Snapshot(
 		return err
 	}
 
-	err = r.restoreIndexes(ctx, oplogOption.nss, cloneNS)
+	if cloneNS.IsSpecified() {
+		err = r.restoreIndexes(ctx, []string{cloneNS.ToNS})
+	} else {
+		err = r.restoreIndexes(ctx, oplogOption.nss)
+	}
 	if err != nil {
 		return errors.Wrap(err, "restore indexes")
 	}
@@ -439,7 +443,11 @@ func (r *Restore) PITR(
 		return err
 	}
 
-	err = r.restoreIndexes(ctx, oplogOption.nss, cloneNS)
+	if cloneNS.IsSpecified() {
+		err = r.restoreIndexes(ctx, []string{cloneNS.ToNS})
+	} else {
+		err = r.restoreIndexes(ctx, oplogOption.nss)
+	}
 	if err != nil {
 		return errors.Wrap(err, "restore indexes")
 	}
@@ -863,7 +871,7 @@ func (r *Restore) RunSnapshot(
 					return nil, err
 				}
 
-				err = r.loadIndexesFrom(bytes.NewReader(data))
+				err = r.loadIndexesFrom(bytes.NewReader(data), cloneNS)
 				if err != nil {
 					return nil, errors.Wrap(err, "load indexes")
 				}
@@ -1007,12 +1015,15 @@ func (r *Restore) restoreUsersAndRoles(ctx context.Context, nss []string) error
 	return nil
 }
 
-func (r *Restore) loadIndexesFrom(rdr io.Reader) error {
+func (r *Restore) loadIndexesFrom(rdr io.Reader, cloneNS snapshot.CloneNS) error {
 	meta, err := archive.ReadMetadata(rdr)
 	if err != nil {
 		return errors.Wrap(err, "read metadata")
 	}
 
+	fromDB, fromColl := cloneNS.SplitFromNS()
+	toDB, toColl := cloneNS.SplitToNS()
+
 	for _, ns := range meta.Namespaces {
 		var md mongorestore.Metadata
 		err := bson.UnmarshalExtJSON([]byte(ns.Metadata), true, &md)
@@ -1021,7 +1032,11 @@ func (r *Restore) loadIndexesFrom(rdr io.Reader) error {
 				ns.Database, ns.Collection)
 		}
 
-		r.indexCatalog.AddIndexes(ns.Database, ns.Collection, md.Indexes)
+		if cloneNS.IsSpecified() && ns.Database == fromDB && ns.Collection == fromColl {
+			r.indexCatalog.AddIndexes(toDB, toColl, md.Indexes)
+		} else {
+			r.indexCatalog.AddIndexes(ns.Database, ns.Collection, md.Indexes)
+		}
 
 		simple := true
 		if md.Options != nil {
@@ -1048,7 +1063,7 @@ func (r *Restore) loadIndexesFrom(rdr io.Reader) error {
 	return nil
 }
 
-func (r *Restore) restoreIndexes(ctx context.Context, nss []string, cloneNS snapshot.CloneNS) error {
+func (r *Restore) restoreIndexes(ctx context.Context, nss []string) error {
 	r.log.Debug("building indexes up")
 
 	isSelected := util.MakeSelectedPred(nss)
@@ -1073,32 +1088,24 @@ func (r *Restore) restoreIndexes(ctx context.Context, nss []string, cloneNS snap
 		}
 
 		var indexNames []string
-		var targetDB, targetColl string
 		for _, index := range indexes {
-			if cloneNS.IsSpecified() && ns.String() == cloneNS.FromNS {
-				// override index's ns for the collection cloning
-				targetDB, targetColl = util.ParseNS(cloneNS.ToNS)
-				index.Options["ns"] = cloneNS.ToNS
-			} else {
-				targetDB, targetColl = ns.DB, ns.Collection
-				index.Options["ns"] = ns.DB + "." + ns.Collection
-			}
+			index.Options["ns"] = ns.DB + "." + ns.Collection
 			indexNames = append(indexNames, index.Options["name"].(string))
 			// remove the index version, forcing an update
 			delete(index.Options, "v")
 		}
 
 		rawCommand := bson.D{
-			{"createIndexes", targetColl},
+			{"createIndexes", ns.Collection},
 			{"indexes", indexes},
 			{"ignoreUnknownIndexOptions", true},
 		}
 
 		r.log.Info("restoring indexes for %s.%s: %s",
-			targetDB, targetColl, strings.Join(indexNames, ", "))
-		err := r.nodeConn.Database(targetDB).RunCommand(ctx, rawCommand).Err()
+			ns.DB, ns.Collection, strings.Join(indexNames, ", "))
+		err := r.nodeConn.Database(ns.DB).RunCommand(ctx, rawCommand).Err()
 		if err != nil {
-			return errors.Wrapf(err, "createIndexes for %s.%s", targetDB, targetColl)
+			return errors.Wrapf(err, "createIndexes for %s.%s", ns.DB, ns.Collection)
 		}
 	}
 

pbm/snapshot/restore.go

@@ -3,6 +3,7 @@ package snapshot
 import (
 	"io"
 	"runtime"
+	"strings"
 
 	"github.com/mongodb/mongo-tools/common/options"
 	"github.com/mongodb/mongo-tools/mongorestore"
@@ -42,17 +43,29 @@ var ExcludeFromRestore = []string{
 
 type restorer struct{ *mongorestore.MongoRestore }
 
-// CloneNS contains clone from/to info for cloning NS use case
+// CloneNS contains clone from/to info for cloning NS use case.
 type CloneNS struct {
 	FromNS string
 	ToNS   string
 }
 
-// IsSpecified returns true in case of cloning use case
+// IsSpecified returns true in case of cloning use case.
 func (c *CloneNS) IsSpecified() bool {
 	return c.FromNS != "" && c.ToNS != ""
 }
 
+// SplitFromNS breaks cloning-from namespace to database & collection pair.
+func (c *CloneNS) SplitFromNS() (string, string) {
+	db, coll, _ := strings.Cut(c.FromNS, ".")
+	return db, coll
+}
+
+// SplitToNS breaks cloning-to namespace to database & collection pair.
+func (c *CloneNS) SplitToNS() (string, string) {
+	db, coll, _ := strings.Cut(c.ToNS, ".")
+	return db, coll
+}
+
 func NewRestore(uri string,
 	cfg *config.Config,
 	cloneNS CloneNS,