lib/storage: populate partition names from both small and big directories

Certain partition directories may be missing after restoring from backups
if they had no data. Re-create such directories on start.

lib/storage/table.go

@@ -436,27 +436,17 @@ func (tb *table) PutPartitions(ptws []*partitionWrapper) {
 }
 
 func openPartitions(smallPartitionsPath, bigPartitionsPath string, getDeletedMetricIDs func() *uint64set.Set) ([]*partition, error) {
-	smallD, err := os.Open(smallPartitionsPath)
+	// Certain partition directories in either `big` or `small` dir may be missing
-	if err != nil {
+	// after restoring from backup. So populate partition names from both dirs.
-		return nil, fmt.Errorf("cannot open directory with small partitions %q: %s", smallPartitionsPath, err)
+	ptNames := make(map[string]bool)
+	if err := populatePartitionNames(smallPartitionsPath, ptNames); err != nil {
+		return nil, err
 	}
-	defer fs.MustClose(smallD)
+	if err := populatePartitionNames(bigPartitionsPath, ptNames); err != nil {
-
+		return nil, err
-	fis, err := smallD.Readdir(-1)
-	if err != nil {
-		return nil, fmt.Errorf("cannot read directory with small partitions %q: %s", smallPartitionsPath, err)
 	}
 	var pts []*partition
-	for _, fi := range fis {
+	for ptName := range ptNames {
-		if !fs.IsDirOrSymlink(fi) {
-			// Skip non-directories
-			continue
-		}
-		ptName := fi.Name()
-		if ptName == "snapshots" {
-			// Skipe directory with snapshots
-			continue
-		}
 		smallPartsPath := smallPartitionsPath + "/" + ptName
 		bigPartsPath := bigPartitionsPath + "/" + ptName
 		pt, err := openPartition(smallPartsPath, bigPartsPath, getDeletedMetricIDs)
@@ -469,6 +459,32 @@ func openPartitions(smallPartitionsPath, bigPartitionsPath string, getDeletedMet
 	return pts, nil
 }
 
+func populatePartitionNames(partitionsPath string, ptNames map[string]bool) error {
+	d, err := os.Open(partitionsPath)
+	if err != nil {
+		return fmt.Errorf("cannot open directory with partitions %q: %s", partitionsPath, err)
+	}
+	defer fs.MustClose(d)
+
+	fis, err := d.Readdir(-1)
+	if err != nil {
+		return fmt.Errorf("cannot read directory with partitions %q: %s", partitionsPath, err)
+	}
+	for _, fi := range fis {
+		if !fs.IsDirOrSymlink(fi) {
+			// Skip non-directories
+			continue
+		}
+		ptName := fi.Name()
+		if ptName == "snapshots" {
+			// Skip directory with snapshots
+			continue
+		}
+		ptNames[ptName] = true
+	}
+	return nil
+}
+
 func mustClosePartitions(pts []*partition) {
 	for _, pt := range pts {
 		pt.MustClose()