fs storage: remove empty directories after object deletion

pkg/storages/fs/folder.go

@@ -54,13 +54,46 @@ func (folder *Folder) ListFolder() (objects []storage.Object, subFolders []stora
 
 func (folder *Folder) DeleteObjects(objectsWithRelativePaths []storage.Object) error {
 	for _, object := range objectsWithRelativePaths {
-		err := os.RemoveAll(folder.GetFilePath(object.GetName()))
+		filePath := folder.GetFilePath(object.GetName())
+		err := os.RemoveAll(filePath)
 		if os.IsNotExist(err) {
 			continue
 		}
 		if err != nil {
 			return fmt.Errorf("unable to delete file %q: %w", object.GetName(), err)
 		}
+		if err = folder.cleanupEmptyFolders(path.Dir(filePath)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// cleanupEmptyFolders removes empty parent directories up to the root folder path.
+func (folder *Folder) cleanupEmptyFolders(dirPath string) error {
+	rootPath := path.Clean(folder.rootPath)
+	for {
+		dirPath = path.Clean(dirPath)
+		if dirPath == rootPath || !strings.HasPrefix(dirPath, rootPath) {
+			break
+		}
+		entries, err := os.ReadDir(dirPath)
+		if err != nil {
+			if os.IsNotExist(err) {
+				break
+			}
+			return fmt.Errorf("unable to read directory %q: %w", dirPath, err)
+		}
+		if len(entries) > 0 {
+			break
+		}
+		if err = os.Remove(dirPath); err != nil {
+			if os.IsNotExist(err) {
+				break
+			}
+			return fmt.Errorf("unable to remove empty directory %q: %w", dirPath, err)
+		}
+		dirPath = path.Dir(dirPath)
 	}
 	return nil
 }

pkg/storages/fs/folder_test.go

@@ -3,7 +3,9 @@ package fs
 import (
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/wal-g/wal-g/pkg/storages/storage"
@@ -20,6 +22,65 @@ func TestFSFolder(t *testing.T) {
 	storage.RunFolderTest(st.RootFolder(), t)
 }
 
+func TestDeleteObjectsRemovesEmptyFolders(t *testing.T) {
+	tmpDir := setupTmpDir(t)
+	defer os.RemoveAll(tmpDir)
+
+	st, err := ConfigureStorage(tmpDir, nil)
+	assert.NoError(t, err)
+
+	root := st.RootFolder()
+
+	// Create a nested structure: root/a/b/file
+	sub := root.GetSubFolder("a/b")
+	err = sub.PutObject("file", strings.NewReader("data"))
+	assert.NoError(t, err)
+
+	// Verify the directory exists
+	_, err = os.Stat(filepath.Join(tmpDir, "a", "b"))
+	assert.NoError(t, err)
+
+	// Delete the file
+	err = sub.DeleteObjects([]storage.Object{storage.NewLocalObject("file", time.Time{}, 0)})
+	assert.NoError(t, err)
+
+	// Both empty subdirectories should have been cleaned up
+	_, err = os.Stat(filepath.Join(tmpDir, "a", "b"))
+	assert.True(t, os.IsNotExist(err), "empty subdirectory 'b' should have been removed")
+
+	_, err = os.Stat(filepath.Join(tmpDir, "a"))
+	assert.True(t, os.IsNotExist(err), "empty subdirectory 'a' should have been removed")
+
+	// Root directory must still exist
+	_, err = os.Stat(tmpDir)
+	assert.NoError(t, err, "root directory must not be removed")
+}
+
+func TestDeleteObjectsKeepsNonEmptyFolders(t *testing.T) {
+	tmpDir := setupTmpDir(t)
+	defer os.RemoveAll(tmpDir)
+
+	st, err := ConfigureStorage(tmpDir, nil)
+	assert.NoError(t, err)
+
+	root := st.RootFolder()
+
+	// Create two files in the same subdirectory
+	sub := root.GetSubFolder("a/b")
+	err = sub.PutObject("file1", strings.NewReader("data1"))
+	assert.NoError(t, err)
+	err = sub.PutObject("file2", strings.NewReader("data2"))
+	assert.NoError(t, err)
+
+	// Delete only one file
+	err = sub.DeleteObjects([]storage.Object{storage.NewLocalObject("file1", time.Time{}, 0)})
+	assert.NoError(t, err)
+
+	// The subdirectory should still exist because file2 is still there
+	_, err = os.Stat(filepath.Join(tmpDir, "a", "b"))
+	assert.NoError(t, err, "non-empty subdirectory 'b' should not be removed")
+}
+
 func setupTmpDir(t *testing.T) string {
 	cwd, err := filepath.Abs("./")
 	if err != nil {