batch delete orphan data by file keys

1 files changed, 29 insertions, 18 deletions

diff --git a/weed/server/volume_grpc_batch_delete.go b/weed/server/volume_grpc_batch_delete.go
index fdb7937d2..501964191 100644
--- a/weed/server/volume_grpc_batch_delete.go
+++ b/weed/server/volume_grpc_batch_delete.go
@@ -8,6 +8,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/storage/types"
 )
 
 func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) {
@@ -28,16 +29,34 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
 
 		n := new(needle.Needle)
 		volumeId, _ := needle.NewVolumeId(vid)
-		n.ParsePath(id_cookie)
-
-		cookie := n.Cookie
-		if _, err := vs.store.ReadVolumeNeedle(volumeId, n); err != nil {
-			resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
-				FileId: fid,
-				Status: http.StatusNotFound,
-				Error:  err.Error(),
-			})
-			continue
+		if req.SkipCookieCheck {
+			n.Id, err = types.ParseNeedleId(id_cookie)
+			if err != nil {
+				resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
+					FileId: fid,
+					Status: http.StatusBadRequest,
+					Error:  err.Error()})
+				continue
+			}
+		} else {
+			n.ParsePath(id_cookie)
+			cookie := n.Cookie
+			if _, err := vs.store.ReadVolumeNeedle(volumeId, n); err != nil {
+				resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
+					FileId: fid,
+					Status: http.StatusNotFound,
+					Error:  err.Error(),
+				})
+				continue
+			}
+			if n.Cookie != cookie {
+				resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
+					FileId: fid,
+					Status: http.StatusBadRequest,
+					Error:  "File Random Cookie does not match.",
+				})
+				break
+			}
 		}
 
 		if n.IsChunkedManifest() {
@@ -49,14 +68,6 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
 			continue
 		}
 
-		if n.Cookie != cookie {
-			resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
-				FileId: fid,
-				Status: http.StatusBadRequest,
-				Error:  "File Random Cookie does not match.",
-			})
-			break
-		}
 		n.LastModified = now
 		if size, err := vs.store.DeleteVolumeNeedle(volumeId, n); err != nil {
 			resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{