Move all logic to operation

src/Disks/ObjectStorages/DiskObjectStorage.cpp

@@ -376,14 +376,9 @@ void DiskObjectStorage::removeSharedFileIfExists(const String & path, bool delet
 void DiskObjectStorage::removeSharedRecursive(
     const String & path, bool keep_all_batch_data, const NameSet & file_names_remove_metadata_only)
 {
-    /// At most remove_shared_recursive_file_limit files are removed in one transaction
+    auto transaction = createObjectStorageTransaction();
-    /// Retry until all of them are removed
+    transaction->removeSharedRecursive(path, keep_all_batch_data, file_names_remove_metadata_only);
-    while (exists(path))
+    transaction->commit();
-    {
-        auto transaction = createObjectStorageTransaction();
-        transaction->removeSharedRecursive(path, keep_all_batch_data, file_names_remove_metadata_only);
-        transaction->commit();
-    }
 }
 
 bool DiskObjectStorage::tryReserve(UInt64 bytes)

src/Disks/ObjectStorages/DiskObjectStorageTransaction.cpp

@@ -326,11 +326,46 @@ struct RemoveRecursiveObjectStorageOperation final : public IDiskObjectStorageOp
         return limit > 0 && objects_to_remove_by_path.size() == limit;
     }
 
+    void removeObjects()
+    {
+        if (!keep_all_batch_data)
+        {
+            std::vector<String> total_removed_paths;
+            total_removed_paths.reserve(objects_to_remove_by_path.size());
+
+            StoredObjects remove_from_remote;
+            for (auto && [local_path, objects_to_remove] : objects_to_remove_by_path)
+            {
+                chassert(!file_names_remove_metadata_only.contains(local_path));
+                if (objects_to_remove.unlink_outcome->num_hardlinks == 0)
+                {
+                    std::move(objects_to_remove.objects.begin(), objects_to_remove.objects.end(), std::back_inserter(remove_from_remote));
+                    total_removed_paths.push_back(local_path);
+                }
+            }
+
+            /// Read comment inside RemoveObjectStorageOperation class
+            /// TL;DR Don't pay any attention to 404 status code
+            object_storage.removeObjectsIfExist(remove_from_remote);
+
+            objects_to_remove_by_path.clear();
+
+            LOG_DEBUG(
+                getLogger("RemoveRecursiveObjectStorageOperation"),
+                "Recursively remove path {}: "
+                "metadata and objects were removed for [{}], "
+                "only metadata were removed for [{}].",
+                path,
+                boost::algorithm::join(total_removed_paths, ", "),
+                boost::algorithm::join(file_names_remove_metadata_only, ", "));
+        }
+    }
+
     void removeMetadataRecursive(MetadataTransactionPtr tx, const std::string & path_to_remove)
     {
         checkStackSize(); /// This is needed to prevent stack overflow in case of cyclic symlinks.
         if (checkLimitReached())
-            return;
+            removeObjects();
 
         if (metadata_storage.isFile(path_to_remove))
         {
@@ -373,13 +408,11 @@ struct RemoveRecursiveObjectStorageOperation final : public IDiskObjectStorageOp
         {
             for (auto it = metadata_storage.iterateDirectory(path_to_remove); it->isValid(); it->next())
             {
-                if (checkLimitReached())
-                    return;
                 removeMetadataRecursive(tx, it->path());
+                if (checkLimitReached())
+                    removeObjects();
             }
-            /// Do not delete in case directory contains >= limit files
+            tx->removeDirectory(path_to_remove);
-            if (objects_to_remove_by_path.size() < limit)
-                tx->removeDirectory(path_to_remove);
         }
     }
 
@@ -396,35 +429,7 @@ struct RemoveRecursiveObjectStorageOperation final : public IDiskObjectStorageOp
 
     void finalize() override
     {
-        if (!keep_all_batch_data)
+        removeObjects();
-        {
-            std::vector<String> total_removed_paths;
-            total_removed_paths.reserve(objects_to_remove_by_path.size());
-
-            StoredObjects remove_from_remote;
-            for (auto && [local_path, objects_to_remove] : objects_to_remove_by_path)
-            {
-                chassert(!file_names_remove_metadata_only.contains(local_path));
-                if (objects_to_remove.unlink_outcome->num_hardlinks == 0)
-                {
-                    std::move(objects_to_remove.objects.begin(), objects_to_remove.objects.end(), std::back_inserter(remove_from_remote));
-                    total_removed_paths.push_back(local_path);
-                }
-            }
-
-            /// Read comment inside RemoveObjectStorageOperation class
-            /// TL;DR Don't pay any attention to 404 status code
-            object_storage.removeObjectsIfExist(remove_from_remote);
-
-            LOG_DEBUG(
-                getLogger("RemoveRecursiveObjectStorageOperation"),
-                "Recursively remove path {}: "
-                "metadata and objects were removed for [{}], "
-                "only metadata were removed for [{}].",
-                path,
-                boost::algorithm::join(total_removed_paths, ", "),
-                boost::algorithm::join(file_names_remove_metadata_only, ", "));
-        }
     }
 };