Stitched blobs support different sized component blobs

ambry-api/src/main/java/com.github.ambry/config/RouterConfig.java

@@ -193,6 +193,13 @@ public class RouterConfig {
   @Default("5")
   public final short routerBlobidCurrentVersion;
 
+  /**
+   * The version to use for new metadata blobs.
+   */
+  @Config("router.metadata.content.version")
+  @Default("2")
+  public final short routerMetadataContentVersion;
+
   /**
    * The KeyManagementServiceFactory that will be used to fetch {@link com.github.ambry.router.KeyManagementService}
    */
@@ -292,6 +299,8 @@ public RouterConfig(VerifiableProperties verifiableProperties) {
     routerBlobidCurrentVersion =
         verifiableProperties.getShortFromAllowedValues("router.blobid.current.version", (short) 6,
             new Short[]{1, 2, 3, 4, 5, 6});
+    routerMetadataContentVersion =
+        verifiableProperties.getShortFromAllowedValues("router.metadata.content.version", (short) 2, new Short[]{2, 3});
     routerKeyManagementServiceFactory =
         verifiableProperties.getString("router.key.management.service.factory", DEFAULT_KMS_FACTORY);
     routerCryptoServiceFactory =

ambry-messageformat/src/main/java/com.github.ambry.messageformat/CompositeBlobInfo.java

@@ -15,17 +15,25 @@
 package com.github.ambry.messageformat;
 
 import com.github.ambry.store.StoreKey;
+import com.github.ambry.utils.Pair;
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 
 
 /**
  * This class holds information about a composite blob parsed from the metadata blob. It contains the chunk size,
  * total composite blob size, and a list of keys for the blob's data chunks.
  */
 public class CompositeBlobInfo {
+
   private final int chunkSize;
   private final long totalSize;
-  private final List<StoreKey> keys;
+  private final List<Long> offsets = new ArrayList<>();
+  private final List<ChunkMetadata> chunkMetadataList = new ArrayList<>();
+  private final short metadataContentVersion;
 
   /**
    * Construct a {@link CompositeBlobInfo} object.
@@ -35,8 +43,81 @@ public class CompositeBlobInfo {
    */
   public CompositeBlobInfo(int chunkSize, long totalSize, List<StoreKey> keys) {
     this.chunkSize = chunkSize;
+    if (chunkSize < 1L) {
+      throw new IllegalArgumentException("chunkSize cannot be 0 or less");
+    }
+    if (keys == null || keys.isEmpty()) {
+      throw new IllegalArgumentException("keys should not be null or empty");
+    }
+    long rightAmountOfKeys = totalSize / chunkSize + (totalSize % chunkSize == 0 ? 0 : 1);
+    if (rightAmountOfKeys != keys.size()) {
+      throw new IllegalArgumentException("keys should contain " + rightAmountOfKeys + " keys, not " + keys.size());
+    }
     this.totalSize = totalSize;
-    this.keys = keys;
+    metadataContentVersion = MessageFormatRecord.Metadata_Content_Version_V2;
+    long last = 0;
+    for (StoreKey key : keys) {
+      offsets.add(last);
+      if (totalSize - last < 1L) {
+        throw new IllegalArgumentException("CompositeBlobInfo can't be composed of blobs with size 0 or less");
+      }
+      chunkMetadataList.add(new ChunkMetadata(key, last, Math.min(chunkSize, totalSize - last)));
+      last += chunkSize;
+    }
+  }
+
+  /**
+   * Construct a {@link CompositeBlobInfo} object.
+   * @param keysAndContentSizes list of store keys and the size of the data content they reference
+   */
+  public CompositeBlobInfo(List<Pair<StoreKey, Long>> keysAndContentSizes) {
+    if (keysAndContentSizes == null || keysAndContentSizes.isEmpty()) {
+      throw new IllegalArgumentException("keysAndContentSizes should not be null or empty");
+    }
+    this.chunkSize = -1;
+    long last = 0;
+    metadataContentVersion = MessageFormatRecord.Metadata_Content_Version_V3;
+    for (Pair<StoreKey, Long> keyAndContentSize : keysAndContentSizes) {
+      if (keyAndContentSize.getSecond() < 1L) {
+        throw new IllegalArgumentException("CompositeBlobInfo can't be composed of blobs with size 0 or less");
+      }
+      StoreKey key = keyAndContentSize.getFirst();
+      offsets.add(last);
+      chunkMetadataList.add(new ChunkMetadata(key, last, keyAndContentSize.getSecond()));
+      last += keyAndContentSize.getSecond();
+    }
+    this.totalSize = last;
+  }
+
+  /**
+   * Returns the keys (along with their data content size and offset relative to the total
+   * composite blob) of the chunks related to the given byte range of the entire composite
+   * blob
+   * @param start inclusive starting byte index
+   * @param end inclusive ending byte index
+   * @return the list of {@link ChunkMetadata} that lie upon the inclusive range
+   * between 'start' and 'end'
+   */
+  public List<ChunkMetadata> getStoreKeysInByteRange(long start, long end) {
+    if (end < start || start < 0L || end >= totalSize) {
+      throw new IllegalArgumentException(
+          "Bad input parameters, start=" + start + " end=" + end + " totalSize=" + totalSize);
+    }
+    int startIdx = Collections.binarySearch(offsets, start);
+    //binarySearch returns -(insertion point) - 1 if not an exact match, which points to the index of
+    //the first element greater than 'start', or list.size() if all elements in the list are
+    // less than 'start'.
+    if (startIdx < 0) {
+      startIdx = -startIdx - 2; //points to element with offset closest to but less than 'start'
+    }
+
+    int endIdx = Collections.binarySearch(offsets, end);
+    if (endIdx < 0) {
+      endIdx = -endIdx - 2; //points to element with offset closest to but less than 'end'
+    }
+    endIdx++; //List.subList expects an exclusive index
+
+    return chunkMetadataList.subList(startIdx, endIdx);
   }
 
   /**
@@ -60,6 +141,75 @@ public int getChunkSize() {
    * @return A list of {@link StoreKey}s.
    */
   public List<StoreKey> getKeys() {
-    return keys;
+    return new AbstractList<StoreKey>() {
+      @Override
+      public StoreKey get(int index) {
+        return chunkMetadataList.get(index).getStoreKey();
+      }
+
+      @Override
+      public int size() {
+        return chunkMetadataList.size();
+      }
+    };
+  }
+
+  /**
+   * Get the list of keys for the composite blob's data chunks, along with the
+   * key's data content size and offset relative to the total composite blob
+   * @return A list of {@link ChunkMetadata}
+   */
+  public List<ChunkMetadata> getChunkMetadataList() {
+    return chunkMetadataList;
+  }
+
+  /**
+   * Return the version of the metadata content record that this object was deserialized from
+   * @return the version of the metadata content record that this object was deserialized from
+   */
+  public short getMetadataContentVersion() {
+    return metadataContentVersion;
+  }
+
+  /**
+   * POJO class for holding a store key, the size of the data content it refers to,
+   * and the offset of the data in the larger composite blob
+   */
+  public static class ChunkMetadata {
+    private final StoreKey storeKey;
+    private final long offset;
+    private final long size;
+
+    public ChunkMetadata(StoreKey storeKey, long offset, long size) {
+      this.storeKey = storeKey;
+      this.offset = offset;
+      this.size = size;
+    }
+
+    public StoreKey getStoreKey() {
+      return storeKey;
+    }
+
+    public long getOffset() {
+      return offset;
+    }
+
+    public long getSize() {
+      return size;
+    }
+
+    @Override
+    public boolean equals(Object that) {
+      if (that instanceof ChunkMetadata) {
+        ChunkMetadata thatCM = (ChunkMetadata) that;
+        return storeKey.equals(thatCM.storeKey) && offset == thatCM.offset && size == thatCM.size;
+      }
+      return false;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(storeKey, offset, size);
+    }
   }
 }

ambry-messageformat/src/main/java/com.github.ambry.messageformat/MessageFormatRecord.java

@@ -18,6 +18,7 @@
 import com.github.ambry.utils.ByteBufferInputStream;
 import com.github.ambry.utils.Crc32;
 import com.github.ambry.utils.CrcInputStream;
+import com.github.ambry.utils.Pair;
 import com.github.ambry.utils.Utils;
 import java.io.DataInputStream;
 import java.io.IOException;
@@ -54,6 +55,7 @@ public class MessageFormatRecord {
   public static final short Blob_Version_V1 = 1;
   public static final short Blob_Version_V2 = 2;
   public static final short Metadata_Content_Version_V2 = 2;
+  public static final short Metadata_Content_Version_V3 = 3;
   public static final int Message_Header_Invalid_Relative_Offset = -1;
 
   static short headerVersionToUse = Message_Header_Version_V2;
@@ -1457,7 +1459,7 @@ public static void serializeMetadataContentRecord(ByteBuffer outputBuffer, int c
      * @throws MessageFormatException
      */
     public static CompositeBlobInfo deserializeMetadataContentRecord(DataInputStream stream,
-        StoreKeyFactory storeKeyFactory) throws IOException, MessageFormatException {
+        StoreKeyFactory storeKeyFactory) throws IOException {
       List<StoreKey> keys = new ArrayList<StoreKey>();
       int chunkSize = stream.readInt();
       long totalSize = stream.readLong();
@@ -1469,6 +1471,97 @@ public static CompositeBlobInfo deserializeMetadataContentRecord(DataInputStream
       return new CompositeBlobInfo(chunkSize, totalSize, keys);
     }
   }
+
+  /**
+   *  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
+   * |         |              |            |            |          |            |            |          |
+   * | version |  total size  | # of keys  |  size of   |   key1   |  size of   |    key2    |  ......  |
+   * |(2 bytes)|  (8 bytes)   | (4 bytes)  | key1 blob  |          | key2 blob  |            |  ......  |
+   * |         |              |            | (8 bytes)  |          | (8 bytes)  |            |          |
+   *  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
+   *  version           - The version of the metadata content record
+   *
+   *  total size        - total size of the object this metadata describes.
+   *
+   *  # of keys         - number of keys in the metadata blob
+   *
+   *  size of key1 blob - size of the data referenced by key1
+   *
+   *  key1              - first key to be part of metadata blob
+   *
+   *  size of key2 blob - size of the data referenced by key2
+   *
+   *  key2              - second key to be part of metadata blob
+   *
+   */
+  public static class Metadata_Content_Format_V3 {
+    private static final int NUM_OF_KEYS_FIELD_SIZE_IN_BYTES = 4;
+    private static final int SIZE_OF_BLOB_FIELD_SIZE_IN_BYTES = 8;
+    private static final int TOTAL_SIZE_FIELD_SIZE_IN_BYTES = 8;
+
+    /**
+     * Get the total size of the metadata content record.
+     * @param keySize The size of each key in bytes.
+     * @param numberOfKeys The total number of keys.
+     * @return The total size in bytes.
+     */
+    public static int getMetadataContentSize(int keySize, int numberOfKeys) {
+      return Version_Field_Size_In_Bytes + TOTAL_SIZE_FIELD_SIZE_IN_BYTES + NUM_OF_KEYS_FIELD_SIZE_IN_BYTES
+          + numberOfKeys * (keySize + SIZE_OF_BLOB_FIELD_SIZE_IN_BYTES);
+    }
+
+    /**
+     * Serialize a metadata content record.
+     * @param outputBuffer output buffer of the serialized metadata content
+     * @param totalSize total size of the blob data content
+     * @param keysAndContentSizes list of data blob keys referenced by the metadata
+     *                            blob along with the data content sizes of each data blob
+     */
+    public static void serializeMetadataContentRecord(ByteBuffer outputBuffer, long totalSize,
+        List<Pair<StoreKey, Long>> keysAndContentSizes) {
+      int keySize = keysAndContentSizes.get(0).getFirst().sizeInBytes();
+      outputBuffer.putShort(Metadata_Content_Version_V3);
+      outputBuffer.putLong(totalSize);
+      outputBuffer.putInt(keysAndContentSizes.size());
+      long sum = 0;
+      for (Pair<StoreKey, Long> keyAndContentSize : keysAndContentSizes) {
+        if (keyAndContentSize.getFirst().sizeInBytes() != keySize) {
+          throw new IllegalArgumentException("Keys are not of same size");
+        }
+        outputBuffer.putLong(keyAndContentSize.getSecond());
+        outputBuffer.put(keyAndContentSize.getFirst().toBytes());
+        sum += keyAndContentSize.getSecond();
+      }
+      if (sum != totalSize) {
+        throw new IllegalArgumentException("Key content sizes do not equal total size");
+      }
+    }
+
+    /**
+     * Deserialize a metadata content record from a stream.
+     * @param stream The stream to read the serialized record from.
+     * @param storeKeyFactory The factory to use for parsing keys in the serialized metadata content record.
+     * @return A {@link CompositeBlobInfo} object with the chunk size and list of keys from the record.
+     * @throws IOException
+     */
+    public static CompositeBlobInfo deserializeMetadataContentRecord(DataInputStream stream,
+        StoreKeyFactory storeKeyFactory) throws IOException {
+      List<Pair<StoreKey, Long>> keysAndContentSizes = new ArrayList<>();
+      long totalSize = stream.readLong();
+      long sum = 0;
+      int numberOfKeys = stream.readInt();
+      for (int i = 0; i < numberOfKeys; i++) {
+        long contentSize = stream.readLong();
+        StoreKey storeKey = storeKeyFactory.getStoreKey(stream);
+        keysAndContentSizes.add(new Pair<>(storeKey, contentSize));
+        sum += contentSize;
+      }
+      if (sum != totalSize) {
+        throw new IllegalArgumentException("Key content sizes do not equal total size");
+      }
+      return new CompositeBlobInfo(keysAndContentSizes);
+    }
+  }
 }
 
 class DeserializedBlobProperties {

ambry-messageformat/src/main/java/com.github.ambry.messageformat/MetadataContentSerDe.java

@@ -16,24 +16,27 @@
 import com.github.ambry.store.StoreKey;
 import com.github.ambry.store.StoreKeyFactory;
 import com.github.ambry.utils.ByteBufferInputStream;
+import com.github.ambry.utils.Pair;
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.stream.Collectors;
 
 
 /**
  * A class to serialize and deserialize MetadataContent which forms the content of a Metadata Blob.
  */
 public class MetadataContentSerDe {
+
   /**
    * Serialize the input list of keys that form the metadata content.
    * @param chunkSize the size of the intermediate data chunks for the object this metadata describes.
    * @param totalSize the total size of the object this metadata describes.
    * @param keys the input list of keys that form the metadata content.
    * @return a ByteBuffer containing the serialized output.
    */
-  public static ByteBuffer serializeMetadataContent(int chunkSize, long totalSize, List<StoreKey> keys) {
+  public static ByteBuffer serializeMetadataContentV2(int chunkSize, long totalSize, List<StoreKey> keys) {
     int bufSize =
         MessageFormatRecord.Metadata_Content_Format_V2.getMetadataContentSize(keys.get(0).sizeInBytes(), keys.size());
     ByteBuffer outputBuf = ByteBuffer.allocate(bufSize);
@@ -42,6 +45,21 @@ public static ByteBuffer serializeMetadataContent(int chunkSize, long totalSize,
     return outputBuf;
   }
 
+  /**
+   * Serialize the input list of keys with data content sizes that form the metadata content.
+   * @param totalSize the total size of the object this metadata describes.
+   * @param keysAndContentSizes the input list of keys and related data content sizes that form the metadata content.
+   * @return a ByteBuffer containing the serialized output.
+   */
+  public static ByteBuffer serializeMetadataContentV3(long totalSize, List<Pair<StoreKey, Long>> keysAndContentSizes) {
+    int bufSize = MessageFormatRecord.Metadata_Content_Format_V3.getMetadataContentSize(
+        keysAndContentSizes.get(0).getFirst().sizeInBytes(), keysAndContentSizes.size());
+    ByteBuffer outputBuf = ByteBuffer.allocate(bufSize);
+    MessageFormatRecord.Metadata_Content_Format_V3.serializeMetadataContentRecord(outputBuf, totalSize,
+        keysAndContentSizes);
+    return outputBuf;
+  }
+
   /**
    * Deserialize the serialized metadata content in the input ByteBuffer using the given {@link StoreKeyFactory} as a
    * reference.
@@ -58,6 +76,9 @@ public static CompositeBlobInfo deserializeMetadataContentRecord(ByteBuffer buf,
       case MessageFormatRecord.Metadata_Content_Version_V2:
         return MessageFormatRecord.Metadata_Content_Format_V2.deserializeMetadataContentRecord(
             new DataInputStream(new ByteBufferInputStream(buf)), storeKeyFactory);
+      case MessageFormatRecord.Metadata_Content_Version_V3:
+        return MessageFormatRecord.Metadata_Content_Format_V3.deserializeMetadataContentRecord(
+            new DataInputStream(new ByteBufferInputStream(buf)), storeKeyFactory);
       default:
         throw new MessageFormatException("Unknown version encountered for MetadataContent: " + version,
             MessageFormatErrorCodes.Unknown_Format_Version);