Implement deleteSourceObjects in Storage.compose

Modified Storage.ComposeRequest to include the optional deleteSourceObjects parameter.
Added DELETE_SOURCE_OBJECTS option to StorageRpc.
Updated StorageImpl, GrpcStorageImpl, and HttpStorageRpc to propagate the new option to the backend.
Added unit and integration tests to verify the implementation.

Co-authored-by: nidhiii-27 <224584462+nidhiii-27@users.noreply.github.com>

Author: google-labs-jules[bot]

google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java

@@ -645,6 +645,9 @@ public Blob compose(ComposeRequest composeRequest) {
         .forEach(builder::addSourceObjects);
     final Object target = codecs.blobInfo().encode(composeRequest.getTarget());
     builder.setDestination(target);
+    if (composeRequest.deleteSourceObjects()) {
+      builder.setDeleteSourceObjects(true);
+    }
     ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build();
     GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
     return retrier.run(

google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java

@@ -3185,6 +3185,7 @@ class ComposeRequest implements Serializable {
     private final List<SourceBlob> sourceBlobs;
     private final BlobInfo target;
     private final List<BlobTargetOption> targetOptions;
+    private final boolean deleteSourceObjects;
 
     private transient Opts<ObjectTargetOpt> targetOpts;
 
@@ -3222,6 +3223,7 @@ public static class Builder {
       private final Set<BlobTargetOption> targetOptions = new LinkedHashSet<>();
       private BlobInfo target;
       private Opts<ObjectTargetOpt> opts = Opts.empty();
+      private boolean deleteSourceObjects;
 
       /** Add source blobs for compose operation. */
       public Builder addSource(Iterable<String> blobs) {
@@ -3265,6 +3267,16 @@ public Builder setTargetOptions(Iterable<BlobTargetOption> options) {
         return this;
       }
 
+      /**
+       * Sets whether to delete the source objects after the compose operation.
+       *
+       * @return the builder
+       */
+      public Builder setDeleteSourceObjects(boolean deleteSourceObjects) {
+        this.deleteSourceObjects = deleteSourceObjects;
+        return this;
+      }
+
       /** Creates a {@code ComposeRequest} object. */
       public ComposeRequest build() {
         checkArgument(!sourceBlobs.isEmpty());
@@ -3280,6 +3292,7 @@ private ComposeRequest(Builder builder) {
       // keep targetOptions for serialization even though we will read targetOpts
       targetOptions = ImmutableList.copyOf(builder.targetOptions);
       targetOpts = builder.opts.prepend(Opts.unwrap(targetOptions).resolveFrom(target));
+      deleteSourceObjects = builder.deleteSourceObjects;
     }
 
     /** Returns compose operation's source blobs. */
@@ -3297,6 +3310,11 @@ public List<BlobTargetOption> getTargetOptions() {
       return targetOptions;
     }
 
+    /** Returns whether to delete the source objects after the compose operation. */
+    public boolean deleteSourceObjects() {
+      return deleteSourceObjects;
+    }
+
     @InternalApi
     Opts<ObjectTargetOpt> getTargetOpts() {
       return targetOpts;

google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java

@@ -651,7 +651,14 @@ public Blob compose(final ComposeRequest composeRequest) {
     }
     Opts<ObjectTargetOpt> targetOpts = composeRequest.getTargetOpts();
     StorageObject targetPb = codecs.blobInfo().encode(composeRequest.getTarget());
-    Map<StorageRpc.Option, ?> targetOptions = targetOpts.getRpcOptions();
+    final Map<StorageRpc.Option, ?> targetOptions;
+    if (composeRequest.deleteSourceObjects()) {
+      Map<StorageRpc.Option, Object> mutableOptions = new HashMap<>(targetOpts.getRpcOptions());
+      mutableOptions.put(StorageRpc.Option.DELETE_SOURCE_OBJECTS, true);
+      targetOptions = Collections.unmodifiableMap(mutableOptions);
+    } else {
+      targetOptions = targetOpts.getRpcOptions();
+    }
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions);
     return run(

google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java

@@ -817,6 +817,9 @@ public StorageObject compose(
       sourceObjects.add(sourceObject);
     }
     request.setSourceObjects(sourceObjects);
+    if (Boolean.TRUE.equals(targetOptions.get(Option.DELETE_SOURCE_OBJECTS))) {
+      request.setDeleteSourceObjects(true);
+    }
     Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_COMPOSE);
     Scope scope = tracer.withSpan(span);
     try {

google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java

@@ -82,6 +82,7 @@ enum Option {
     INCLUDE_TRAILING_DELIMITER("includeTrailingDelimiter"),
     X_UPLOAD_CONTENT_LENGTH("x-upload-content-length"),
     OBJECT_FILTER("objectFilter"),
+    DELETE_SOURCE_OBJECTS("deleteSourceObjects"),
     /**
      * An {@link com.google.common.collect.ImmutableMap ImmutableMap<String, String>} of values
      * which will be set as additional headers on the request.

google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java

@@ -361,6 +361,20 @@ public void composeRequest() throws IOException, ClassNotFoundException {
     }
   }
 
+  @Test
+  public void testComposeRequestSerialization() throws Exception {
+    Storage.ComposeRequest request =
+        Storage.ComposeRequest.newBuilder()
+            .setTarget(BLOB_INFO)
+            .addSource("s1", "s2")
+            .setDeleteSourceObjects(true)
+            .build();
+    Storage.ComposeRequest copy = serializeAndDeserialize(request);
+    assertThat(copy.getTarget()).isEqualTo(request.getTarget());
+    assertThat(copy.getSourceBlobs().size()).isEqualTo(request.getSourceBlobs().size());
+    assertThat(copy.deleteSourceObjects()).isTrue();
+  }
+
   /**
    * Here we override the super classes implementation to remove the "assertNotSame".
    *

google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java

@@ -52,6 +52,8 @@
 import java.security.spec.PKCS8EncodedKeySpec;
 import java.security.spec.X509EncodedKeySpec;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import javax.crypto.spec.SecretKeySpec;
@@ -134,6 +136,12 @@ public class StorageImplMockitoTest {
       Storage.BlobTargetOption.doesNotExist();
   private static final Storage.BlobTargetOption BLOB_TARGET_PREDEFINED_ACL =
       Storage.BlobTargetOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
+  private static final Storage.ComposeRequest COMPOSE_REQUEST =
+      Storage.ComposeRequest.newBuilder()
+          .setTarget(BLOB_INFO1)
+          .addSource(BLOB_NAME2, BLOB_NAME3)
+          .setDeleteSourceObjects(true)
+          .build();
   private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
@@ -1028,6 +1036,47 @@ public void testDeleteNotification() {
     assertEquals(isDeleted, Boolean.TRUE);
   }
 
+  @Test
+  public void testCompose() {
+    List<StorageObject> sources =
+        ImmutableList.of(
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME2).build()),
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME3).build()));
+    StorageObject target = Conversions.json().blobInfo().encode(BLOB_INFO1);
+    Map<StorageRpc.Option, Object> targetOptions = new HashMap<>();
+    targetOptions.put(StorageRpc.Option.DELETE_SOURCE_OBJECTS, true);
+    doReturn(target)
+        .doThrow(UNEXPECTED_CALL_EXCEPTION)
+        .when(storageRpcMock)
+        .compose(sources, target, targetOptions);
+    initializeService();
+    Blob blob = storage.compose(COMPOSE_REQUEST);
+    assertEquals(expectedBlob1, blob);
+  }
+
+  @Test
+  public void testComposeDeleteSourceObjectsFalse() {
+    List<StorageObject> sources =
+        ImmutableList.of(
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME2).build()),
+            Conversions.json().blobInfo().encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME3).build()));
+    StorageObject target = Conversions.json().blobInfo().encode(BLOB_INFO1);
+    Map<StorageRpc.Option, Object> targetOptions = new HashMap<>();
+    doReturn(target)
+        .doThrow(UNEXPECTED_CALL_EXCEPTION)
+        .when(storageRpcMock)
+        .compose(sources, target, targetOptions);
+    initializeService();
+    Storage.ComposeRequest request =
+        Storage.ComposeRequest.newBuilder()
+            .setTarget(BLOB_INFO1)
+            .addSource(BLOB_NAME2, BLOB_NAME3)
+            .setDeleteSourceObjects(false)
+            .build();
+    Blob blob = storage.compose(request);
+    assertEquals(expectedBlob1, blob);
+  }
+
   private void verifyBucketNotification(Notification value) {
     assertNull(value.getNotificationId());
     assertEquals(CUSTOM_ATTRIBUTES, value.getCustomAttributes());

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java

@@ -812,6 +812,34 @@ public void testComposeBlobWithContentType() {
     assertArrayEquals(composedBytes, readBytes);
   }
 
+  @Test
+  public void testComposeBlobDeleteSource() {
+    String baseName = generator.randomObjectName();
+    String sourceBlobName1 = baseName + "-1";
+    String sourceBlobName2 = baseName + "-2";
+    BlobInfo sourceBlob1 = BlobInfo.newBuilder(bucket, sourceBlobName1).build();
+    BlobInfo sourceBlob2 = BlobInfo.newBuilder(bucket, sourceBlobName2).build();
+    storage.create(sourceBlob1, BLOB_BYTE_CONTENT);
+    storage.create(sourceBlob2, BLOB_BYTE_CONTENT);
+
+    String targetBlobName = baseName + "-target";
+    BlobInfo targetBlob = BlobInfo.newBuilder(bucket, targetBlobName).build();
+    ComposeRequest req =
+        ComposeRequest.newBuilder()
+            .addSource(sourceBlobName1, sourceBlobName2)
+            .setTarget(targetBlob)
+            .setDeleteSourceObjects(true)
+            .build();
+    Blob remoteTargetBlob = storage.compose(req);
+    assertNotNull(remoteTargetBlob);
+
+    assertNull(storage.get(bucket.getName(), sourceBlobName1));
+    assertNull(storage.get(bucket.getName(), sourceBlobName2));
+
+    byte[] readBytes = storage.readAllBytes(bucket.getName(), targetBlobName);
+    assertThat(readBytes.length).isEqualTo(BLOB_BYTE_CONTENT.length * 2);
+  }
+
   @Test
   public void testComposeBlobFail() {
     String baseName = generator.randomObjectName();