feat: Add ability to create a PCU Prefix at the object level (#2345)

* feat: Add ability to create a PCU Prefix at the object level

* Allow no prefix and just create the objects at the ultimate object destination

* Add in tests for the prefix and without a prefix

* linter

* Make using an actual prefix private

* linter

* add in  to the directory where the part files will be created

* Remove directory level in ObjectNamePrefix Naming Strategy

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: sydney-munro

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

@@ -462,10 +462,7 @@ String fmtName(String ultimateObjectName, PartRange partRange) {
 
       // encode it to base 64, yielding 22 characters
       String randomKey = B64.encodeToString(bytes);
-      HashCode hashCode =
-          OBJECT_NAME_HASH_FUNCTION.hashString(ultimateObjectName, StandardCharsets.UTF_8);
-      String nameDigest = B64.encodeToString(hashCode.asBytes());
-      return fmtFields(randomKey, nameDigest, partRange.encode());
+      return fmtFields(randomKey, ultimateObjectName, partRange.encode());
     }
 
     abstract String fmtFields(String randomKey, String nameDigest, String partRange);
@@ -523,6 +520,35 @@ public static PartNamingStrategy prefix(String prefixPattern) {
       return new WithPrefix(rand, prefixPattern);
     }
 
+    /**
+     * Strategy in which the end object name is the prefix included and is present on each part and
+     * intermediary compose object.
+     *
+     * <p>General format is
+     *
+     * <pre><code>
+     *   {objectName}-{randomKeyDigest};{objectInfoDigest};{partIndex}.part
+     * </code></pre>
+     *
+     * <p>{@code {objectInfoDigest}} will be fixed for an individual {@link BlobWriteSession}.
+     *
+     * <p><b><i>NOTE:</i></b>The way in which both {@code randomKeyDigest} and {@code
+     * objectInfoDigest} are generated is undefined and subject to change at any time.
+     *
+     * @see #withPartNamingStrategy(PartNamingStrategy)
+     * @since 2.30.2 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public static PartNamingStrategy useObjectNameAsPrefix() {
+      return useObjectNameAsPrefix("");
+    }
+
+    private static PartNamingStrategy useObjectNameAsPrefix(String prefixPattern) {
+      checkNotNull(prefixPattern, "prefixPattern must be non null");
+      SecureRandom rand = new SecureRandom();
+      return new WithObjectLevelPrefix(rand, prefixPattern);
+    }
+
     static final class WithPrefix extends PartNamingStrategy {
       private static final long serialVersionUID = 5709330763161570411L;
 
@@ -534,7 +560,10 @@ private WithPrefix(SecureRandom rand, String prefix) {
       }
 
       @Override
-      protected String fmtFields(String randomKey, String nameDigest, String partRange) {
+      protected String fmtFields(String randomKey, String ultimateObjectName, String partRange) {
+        HashCode hashCode =
+            OBJECT_NAME_HASH_FUNCTION.hashString(ultimateObjectName, StandardCharsets.UTF_8);
+        String nameDigest = B64.encodeToString(hashCode.asBytes());
         return prefix
             + "/"
             + randomKey
@@ -546,6 +575,35 @@ protected String fmtFields(String randomKey, String nameDigest, String partRange
       }
     }
 
+    static final class WithObjectLevelPrefix extends PartNamingStrategy {
+
+      private static final long serialVersionUID = 5157942020618764450L;
+      private final String prefix;
+
+      private WithObjectLevelPrefix(SecureRandom rand, String prefix) {
+        super(rand);
+        // If no prefix is specified we will create the part files under the same directory as the
+        // ultimate object.
+        this.prefix = prefix.isEmpty() ? prefix : prefix + "/";
+      }
+
+      @Override
+      protected String fmtFields(String randomKey, String ultimateObjectName, String partRange) {
+        HashCode hashCode =
+            OBJECT_NAME_HASH_FUNCTION.hashString(ultimateObjectName, StandardCharsets.UTF_8);
+        String nameDigest = B64.encodeToString(hashCode.asBytes());
+        return prefix
+            + ultimateObjectName
+            + "-"
+            + randomKey
+            + FIELD_SEPARATOR
+            + nameDigest
+            + FIELD_SEPARATOR
+            + partRange
+            + ".part";
+      }
+    }
+
     static final class NoPrefix extends PartNamingStrategy {
       private static final long serialVersionUID = 5202415556658566017L;
 

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

@@ -70,6 +70,23 @@ public void partNameStrategy_prefix_stillWorksWithFmtPattern() throws Exception
         () -> assertThat(fmt).startsWith("[%s]/"));
   }
 
+  @Test
+  public void partNameStrategy_objectNamePrefix() throws Exception {
+    // Creating an object level prefix without specifying an additional prefix will append the
+    // object name to the beginning of the part name.
+    PartNamingStrategy strategy = PartNamingStrategy.useObjectNameAsPrefix();
+
+    String fmt = strategy.fmtName("a/b/obj", PartRange.of(1, 96));
+    assertAll(
+        // random digest with prefix to spread over keyspace
+        // digest is 22, objectName is 7, slash is 1
+        () -> assertField(fmt, 0).hasLength(22 + 8),
+        // name digest
+        () -> assertField(fmt, 1).hasLength(22),
+        () -> assertField(fmt, 2).isEqualTo("0001-0096.part"),
+        () -> assertThat(fmt).startsWith("a/b/obj/"));
+  }
+
   private static StringSubject assertField(String fmt, int idx) {
     String[] split = fmt.split(";");
     String s = split[idx];