storage/disk: respect wildcard-replacement in partition validation

It is now allowed to replace a partition like

    /foo/bar

by

    /foo/*

also if multiple wildcards are used.

Caveats:

You cannot add a wildcard partition like /*/*, since it would overlap
the managed "/system/*" partition.

When attempting to go back from /foo/* to /foo/bar, an error is
raised _unconditionally_ -- we could check the existing data, but
currently don't.

Signed-off-by: Stephan Renatus <stephan.renatus@gmail.com>

storage/disk/disk.go

@@ -485,13 +485,29 @@ func (db *Store) validatePartitions(ctx context.Context, txn *badger.Txn, existi
 	removedPartitions := oldPathSet.Diff(newPathSet)
 	addedPartitions := newPathSet.Diff(oldPathSet)
 
-	if len(removedPartitions) > 0 {
+	// It's OK to replace partitions with wildcard partitions that overlap them:
+	// REMOVED: /foo/bar
+	// ADDED:   /foo/*
+	// and the like.
+	replaced := make(pathSet, 0)
+	replacements := make(pathSet, 0)
+	for _, removed := range removedPartitions {
+		for _, added := range addedPartitions {
+			if isMatchedBy(removed, added) {
+				replaced = append(replaced, removed)
+				replacements = append(replacements, added)
+			}
+		}
+	}
+
+	rest := removedPartitions.Diff(replaced)
+	if len(rest) > 0 {
 		return &storage.Error{
 			Code:    storage.InternalErr,
-			Message: fmt.Sprintf("partitions are backwards incompatible (old: %v, new: %v, missing: %v)", oldPathSet, newPathSet, removedPartitions)}
+			Message: fmt.Sprintf("partitions are backwards incompatible (old: %v, new: %v, missing: %v)", oldPathSet, newPathSet, rest)}
 	}
 
-	for _, path := range addedPartitions {
+	for _, path := range addedPartitions.Diff(replacements) {
 		for i := len(path); i > 0; i-- {
 			key, err := db.pm.DataPath2Key(path[:i])
 			if err != nil {

storage/disk/disk_test.go

@@ -282,6 +282,48 @@ func TestDataPartitioningValidation(t *testing.T) {
 		}
 
 		closeFn(ctx, s)
+
+		// switching to wildcard partition
+		s, err = New(ctx, logging.NewNoOpLogger(), nil, Options{Dir: dir, Partitions: []storage.Path{
+			storage.MustParsePath("/foo/*"),
+		}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		closeFn(ctx, s)
+
+		// adding another partition
+		s, err = New(ctx, logging.NewNoOpLogger(), nil, Options{Dir: dir, Partitions: []storage.Path{
+			storage.MustParsePath("/fox/in/the/snow/*"),
+			storage.MustParsePath("/foo/*"),
+		}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		closeFn(ctx, s)
+
+		// switching to a partition with multiple wildcards
+		s, err = New(ctx, logging.NewNoOpLogger(), nil, Options{Dir: dir, Partitions: []storage.Path{
+			storage.MustParsePath("/fox/in/*/*/*"),
+			storage.MustParsePath("/foo/*"),
+		}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		closeFn(ctx, s)
+
+		// there is no going back
+		s, err = New(ctx, logging.NewNoOpLogger(), nil, Options{Dir: dir, Partitions: []storage.Path{
+			storage.MustParsePath("/fox/in/the/snow/*"),
+			storage.MustParsePath("/foo/*"),
+		}})
+		if err == nil || !strings.Contains(err.Error(),
+			"partitions are backwards incompatible (old: [/foo/* /fox/in/*/*/* /system/*], new: [/foo/* /fox/in/the/snow/* /system/*], missing: [/fox/in/*/*/*])",
+		) {
+			t.Fatal(err)
+		}
+		closeFn(ctx, s)
+
 	})
 }
 

storage/disk/paths.go

@@ -111,6 +111,23 @@ func hasPrefixWithWildcard(p, other storage.Path) bool {
 	return true
 }
 
+// isMatchedBy returns true if p starts with other, or is matched by it
+// respecting wildcards _in other_ -- not in p.
+func isMatchedBy(p, other storage.Path) bool {
+	if len(other) != len(p) {
+		return false
+	}
+	for i := range other {
+		if other[i] == pathWildcard {
+			continue
+		}
+		if p[i] != other[i] {
+			return false
+		}
+	}
+	return true
+}
+
 func (ps pathSet) Diff(other pathSet) pathSet {
 	diff := pathSet{}
 	for _, x := range ps {