Support rename dataset version

ambry-account/src/main/java/com/github/ambry/account/CachedAccountService.java

@@ -621,6 +621,36 @@ public void deleteDatasetVersion(String accountName, String containerName, Strin
     }
   }
 
+  @Override
+  public void renameDatasetVersion(String accountName, String containerName, String datasetName, String sourceVersion,
+      String targetVersion) throws AccountServiceException {
+    try {
+      Pair<Short, Short> accountAndContainerIdPair = getAccountAndContainerIdFromName(accountName, containerName);
+      if (mySqlAccountStore == null) {
+        mySqlAccountStore = this.supplier.get();
+      }
+      mySqlAccountStore.renameDatasetVersion(accountAndContainerIdPair.getFirst(),
+          accountAndContainerIdPair.getSecond(), accountName, containerName, datasetName, sourceVersion, targetVersion);
+    } catch (SQLException e) {
+      throw translateSQLException(e);
+    }
+  }
+
+  @Override
+  public void deleteDatasetVersionForDatasetDelete(String accountName, String containerName, String datasetName,
+      String version) throws AccountServiceException {
+    try {
+      Pair<Short, Short> accountAndContainerIdPair = getAccountAndContainerIdFromName(accountName, containerName);
+      if (mySqlAccountStore == null) {
+        mySqlAccountStore = this.supplier.get();
+      }
+      mySqlAccountStore.deleteDatasetVersionForDatasetDelete(accountAndContainerIdPair.getFirst(),
+          accountAndContainerIdPair.getSecond(), datasetName, version);
+    } catch (SQLException e) {
+      throw translateSQLException(e);
+    }
+  }
+
   @Override
   public void updateDatasetVersionTtl(String accountName, String containerName, String datasetName, String version)
       throws AccountServiceException {

ambry-account/src/main/java/com/github/ambry/account/MySqlAccountService.java

@@ -254,12 +254,24 @@ public DatasetVersionRecord getDatasetVersion(String accountName, String contain
     return cachedAccountService.getDatasetVersion(accountName, containerName, datasetName, version);
   }
 
+  @Override
+  public void renameDatasetVersion(String accountName, String containerName, String datasetName, String sourceVersion,
+      String targetVersion) throws AccountServiceException {
+    cachedAccountService.renameDatasetVersion(accountName, containerName, datasetName, sourceVersion, targetVersion);
+  }
+
   @Override
   public void deleteDatasetVersion(String accountName, String containerName, String datasetName, String version)
       throws AccountServiceException {
     cachedAccountService.deleteDatasetVersion(accountName, containerName, datasetName, version);
   }
 
+  @Override
+  public void deleteDatasetVersionForDatasetDelete(String accountName, String containerName, String datasetName,
+      String version) throws AccountServiceException {
+    cachedAccountService.deleteDatasetVersionForDatasetDelete(accountName, containerName, datasetName, version);
+  }
+
   @Override
   public void updateDatasetVersionTtl(String accountName, String containerName, String datasetName,
       String version) throws AccountServiceException {

ambry-account/src/main/java/com/github/ambry/account/mysql/MySqlAccountStore.java

@@ -199,6 +199,25 @@ public synchronized DatasetVersionRecord addDatasetVersion(int accountId, int co
         timeToLiveInSeconds, creationTimeInMs, datasetVersionTtlEnabled, datasetVersionState);
   }
 
+  /**
+   * Rename a dataset version of {@link Dataset}
+   * @param accountId the id for the parent account.
+   * @param containerId the id of the container.
+   * @param accountName the name for the parent account.
+   * @param containerName the name for the container.
+   * @param datasetName the name of the dataset.
+   * @param sourceVersion the source version to rename.
+   * @param targetVersion the target version to rename.
+   * @throws SQLException
+   * @throws AccountServiceException
+   */
+  public synchronized void renameDatasetVersion(int accountId, int containerId, String accountName,
+      String containerName, String datasetName, String sourceVersion, String targetVersion)
+      throws SQLException, AccountServiceException {
+    datasetDao.renameDatasetVersion(accountId, containerId, accountName, containerName, datasetName, sourceVersion,
+        targetVersion);
+  }
+
   public synchronized void updateDatasetVersionState(int accountId, int containerId, String accountName, String containerName,
       String datasetName, String version, DatasetVersionState datasetVersionState)
       throws SQLException, AccountServiceException {
@@ -236,6 +255,11 @@ public synchronized void deleteDatasetVersion(short accountId, short containerId
     datasetDao.deleteDatasetVersion(accountId, containerId, datasetName, version);
   }
 
+  public synchronized void deleteDatasetVersionForDatasetDelete(short accountId, short containerId, String datasetName,
+      String version) throws SQLException, AccountServiceException {
+    datasetDao.deleteDatasetVersionForDatasetDelete(accountId, containerId, datasetName, version);
+  }
+
   /**
    * Update ttl for a version of {@link Dataset}
    * @param accountId the id for the parent account.

ambry-account/src/main/resources/AccountSchema.ddl

@@ -73,6 +73,7 @@ CREATE TABLE IF NOT EXISTS DatasetVersions (
     lastModifiedTime DATETIME(3) NOT NULL,
     delete_ts DATETIME(6) DEFAULT NULL,
     deleted_ts DATETIME(6) DEFAULT NULL,
+    rename_from VARCHAR(25) DEFAULT NULL,
     PRIMARY KEY (accountId, containerId, datasetName, version)
 )
 CHARACTER SET utf8 COLLATE utf8_bin;

ambry-api/src/main/java/com/github/ambry/account/AccountService.java

@@ -304,6 +304,25 @@ default void deleteDatasetVersion(String accountName, String containerName, Stri
     throw new UnsupportedOperationException("This method is not supported");
   }
 
+  default void deleteDatasetVersionForDatasetDelete(String accountName, String containerName, String datasetName,
+      String version) throws AccountServiceException {
+    throw new UnsupportedOperationException("This method is not supported");
+  }
+
+  /**
+   * Rename a dataset version.
+   * @param accountName The name for the parent account.
+   * @param containerName The name for the container.
+   * @param datasetName The name of the dataset.
+   * @param sourceVersion the source version which need to be renamed.
+   * @param targetVersion the target version to rename to.
+   * @throws AccountServiceException
+   */
+  default void renameDatasetVersion(String accountName, String containerName, String datasetName, String sourceVersion,
+      String targetVersion) throws AccountServiceException {
+    throw new UnsupportedOperationException("This method is not supported");
+  }
+
   /**
    * Get all valid dataset versions for dataset deletion.
    * @param accountName The name for the parent account.

ambry-api/src/main/java/com/github/ambry/account/DatasetVersionRecord.java

@@ -27,37 +27,46 @@ public class DatasetVersionRecord {
   private final String version;
   private final long expirationTimeMs;
   private final Long creationTimeMs;
+  private String renameFrom;
+  private static final String NAMED_BLOB_PREFIX = "/named";
+  private static final String SLASH = "/";
+
 
   /**
    * Constructor that takes individual arguments.
-   * @param accountId the id of the parent account.
-   * @param containerId the id of the container.
-   * @param datasetName the name of the dataset.
-   * @param version the version of the dataset.
+   *
+   * @param accountId        the id of the parent account.
+   * @param containerId      the id of the container.
+   * @param datasetName      the name of the dataset.
+   * @param version          the version of the dataset.
    * @param expirationTimeMs the expiration time in milliseconds since epoch, or -1 if the blob should be permanent.
+   * @param renameFrom       the original version which renamed from
    */
   public DatasetVersionRecord(int accountId, int containerId, String datasetName, String version,
-      long expirationTimeMs) {
-    this(accountId, containerId, datasetName, version, expirationTimeMs, null);
+      long expirationTimeMs, String renameFrom) {
+    this(accountId, containerId, datasetName, version, expirationTimeMs, null, renameFrom);
   }
 
   /**
    * Constructor for retention policy support.
-   * @param accountId the id of the parent account.
-   * @param containerId the id of the container.
-   * @param datasetName the name of the dataset.
-   * @param version the version of the dataset.
+   *
+   * @param accountId        the id of the parent account.
+   * @param containerId      the id of the container.
+   * @param datasetName      the name of the dataset.
+   * @param version          the version of the dataset.
    * @param expirationTimeMs the expiration time in milliseconds since epoch, or -1 if the blob should be permanent.
-   * @param creationTimeMs the creation time in milliseconds since epoch for dataset version.
+   * @param creationTimeMs   the creation time in milliseconds since epoch for dataset version.
+   * @param renameFrom       the original version which renamed from
    */
   public DatasetVersionRecord(int accountId, int containerId, String datasetName, String version, long expirationTimeMs,
-      Long creationTimeMs) {
+      Long creationTimeMs, String renameFrom) {
     this.accountId = accountId;
     this.containerId = containerId;
     this.datasetName = datasetName;
     this.version = version;
     this.expirationTimeMs = expirationTimeMs;
     this.creationTimeMs = creationTimeMs;
+    this.renameFrom = renameFrom;
   }
 
   /**
@@ -102,6 +111,30 @@ public long getCreationTimeMs() {
     return creationTimeMs;
   }
 
+  public String getRenameFrom() {
+    return renameFrom;
+  }
+
+  private String getOriginalPath(String accountName, String containerName) {
+    return NAMED_BLOB_PREFIX + SLASH + accountName + SLASH + containerName + SLASH + datasetName + SLASH + version;
+  }
+
+  public String getRenamedPath(String accountName, String containerName) {
+    return NAMED_BLOB_PREFIX + SLASH + accountName + SLASH + containerName + SLASH + datasetName + SLASH + renameFrom;
+  }
+
+  public String getNamedBlobNamePath(String accountName, String containerName) {
+    if (this.renameFrom != null) {
+      return getRenamedPath(accountName, containerName);
+    } else {
+      return getOriginalPath(accountName, containerName);
+    }
+  }
+
+  public void setRenameFrom(String renameFrom) {
+    this.renameFrom = renameFrom;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -113,7 +146,7 @@ public boolean equals(Object o) {
     DatasetVersionRecord record = (DatasetVersionRecord) o;
     return accountId == record.accountId && containerId == record.containerId && Objects.equals(datasetName,
         record.datasetName) && Objects.equals(version, record.version) && expirationTimeMs == record.expirationTimeMs
-        && Objects.equals(creationTimeMs, record.creationTimeMs);
+        && Objects.equals(creationTimeMs, record.creationTimeMs) && Objects.equals(renameFrom, record.renameFrom);
   }
 
   @Override

ambry-api/src/main/java/com/github/ambry/frontend/DatasetVersionPath.java

@@ -32,6 +32,13 @@ public class DatasetVersionPath {
   private final String containerName;
   private final String datasetName;
   private final String version;
+  private final String targetVersion;
+  public static final String RENAME = "RENAME";
+
+  public static final String OP = "op";
+  public static final String TARGET_VERSION = "targetVersion";
+
+
 
   public static DatasetVersionPath parse(RequestPath requestPath, Map<String, Object> args)
       throws RestServiceException {
@@ -50,6 +57,7 @@ public static DatasetVersionPath parse(String path, Map<String, Object> args) th
     Objects.requireNonNull(path, "path should not be null");
     Objects.requireNonNull(args, "args should not be null");
     boolean isListRequest = RestUtils.getBooleanHeader(args, ENABLE_DATASET_VERSION_LISTING, false);
+    boolean isRenameRequest = "RENAME".equals(RestUtils.getHeader(args, OP, false));
     path = path.startsWith("/") ? path.substring(1) : path;
     String[] splitPath = path.split("/");
     int expectedMinimumSegments = isListRequest ? 4 : 5;
@@ -65,25 +73,33 @@ public static DatasetVersionPath parse(String path, Map<String, Object> args) th
     //if isListRequest == false, the path format should be /named/<account_name>/<container_name>/<dataset_name>/<version>
     if (isListRequest) {
       datasetName = String.join("/", Arrays.copyOfRange(splitPath, 3, splitPath.length));
-      return new DatasetVersionPath(accountName, containerName, datasetName, null);
+      return new DatasetVersionPath(accountName, containerName, datasetName, null, null);
+    }
+    String targetVersion = null;
+    if (isRenameRequest) {
+      targetVersion = RestUtils.getHeader(args, TARGET_VERSION, true);
     }
     datasetName = String.join("/", Arrays.copyOfRange(splitPath, 3, splitPath.length - 1));
     String version = splitPath[splitPath.length - 1];
-    return new DatasetVersionPath(accountName, containerName, datasetName, version);
+    return new DatasetVersionPath(accountName, containerName, datasetName, version, targetVersion);
   }
 
   /**
    * Construct a {@link DatasetVersionPath}
-   * @param accountName name of the parent account.
+   *
+   * @param accountName   name of the parent account.
    * @param containerName name of the container.
-   * @param datasetName name of the dataset.
-   * @param version the version of the dataset.
+   * @param datasetName   name of the dataset.
+   * @param version       the version of the dataset.
+   * @param targetVersion the target version for copy version API.
    */
-  private DatasetVersionPath(String accountName, String containerName, String datasetName, String version) {
+  private DatasetVersionPath(String accountName, String containerName, String datasetName, String version,
+      String targetVersion) {
     this.accountName = accountName;
     this.containerName = containerName;
     this.datasetName = datasetName;
     this.version = version;
+    this.targetVersion = targetVersion;
   }
 
   /**
@@ -113,4 +129,6 @@ public String getDatasetName() {
   public String getVersion() {
     return version;
   }
+
+  public String getTargetVersion() { return targetVersion; }
 }

ambry-api/src/main/java/com/github/ambry/protocol/DatasetVersionState.java

@@ -16,5 +16,6 @@
 
 public enum DatasetVersionState {
   IN_PROGRESS,
-  READY
+  READY,
+  RENAMED
 }

ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java

@@ -198,6 +198,10 @@ public static final class Headers {
      * for put or get dataset request; long; version of dataset.
      */
     public final static String TARGET_DATASET_VERSION = "x-ambry-target-dataset-version";
+    /**
+     * The original named blob name for a certain dataset version.
+     */
+    public final static String TARGET_DATASET_ORIGINAL_VERSION = "x-ambry-dataset-original-version";
     /**
      * The dataset version expiration time.
      */
@@ -478,6 +482,12 @@ public static final class InternalKeys {
      */
     public static final String REQUEST_PATH = KEY_PREFIX + "request-path";
 
+    /**
+     * The internal header to determine if the delete request is coming from a dataset deletion.
+     */
+    public static final String DATASET_DELETE_ENABLED = KEY_PREFIX + "dataset-delete-enabled";
+
+
     /**
      * To be set to {@code true} if failures reason should be attached to frontend responses.
      */

ambry-frontend/src/integration-test/java/com/github/ambry/frontend/FrontendIntegrationTest.java

@@ -297,6 +297,33 @@ public void datasetTest() throws Exception {
     doDeleteDatasetAndVerify(refAccount.getName(), namedBlobOptionalContainer.getName(), datasetList);
     //After delete, it should have an empty list.
     doListDatasetAndVerify(refAccount.getName(), namedBlobOptionalContainer.getName(), new ArrayList<>());
+
+  }
+
+  @Ignore
+  @Test
+  public void datasetRenameTest() throws Exception {
+    Account refAccount = ACCOUNT_SERVICE.createAndAddRandomAccount();
+    Container namedBlobOptionalContainer =
+        new ContainerBuilder((short) 11, "optional", Container.ContainerStatus.ACTIVE, "",
+            refAccount.getId()).setNamedBlobMode(Container.NamedBlobMode.OPTIONAL).build();
+    ACCOUNT_SERVICE.updateContainers(refAccount.getName(), Arrays.asList(namedBlobOptionalContainer));
+    String contentType = "application/octet-stream";
+    String ownerId = "datasetTest";
+    int contentSize = 100;
+    List<Dataset> datasetList = addSemanticLongDataset(refAccount, namedBlobOptionalContainer, null);
+    List<Pair<String, String>> allDatasetVersions = new ArrayList<>();
+    List<Pair<String, String>> datasetVersionsFromCopy = doCopyDatasetTestAndVerify(datasetList, contentType, ownerId,
+        contentSize);
+    allDatasetVersions.addAll(datasetVersionsFromCopy);
+    //Test List dataset version
+    List<Pair<String, String>> allDatasetVersionPairs = doListDatasetVersionAndVerify(datasetList, allDatasetVersions);
+    doDatasetUpdateTtlAndVerify(refAccount.getName(), namedBlobOptionalContainer.getName(), allDatasetVersionPairs,
+        contentSize, contentType, ownerId);
+    //Test delete
+    doDeleteDatasetVersionAndVerify(refAccount.getName(), namedBlobOptionalContainer.getName(), allDatasetVersionPairs);
+    //After delete, it should have an empty list.
+    doListDatasetVersionAndVerify(datasetList, new ArrayList<>());
   }
 
   @Ignore