1. fixed getLatestTimeStamp bug which takes the null timestamp over t…

…he non-null.

2. Fixed the updateDatasetVersionTtl only updated delete_ts issue.

ambry-account/src/integration-test/java/com/github/ambry/account/MySqlAccountServiceIntegrationTest.java

@@ -952,6 +952,61 @@ public void testAddGetAndUpdateDatasets() throws Exception {
     verifyDatasetProperties(dataset, datasetFromMysql);
   }
 
+  /**
+   * Test the Dataset deletion in the data migration.
+   * 1. if deleted_ts is null but delete_ts is expired, should return deleted.
+   * 2. if deleted_ts and delete_ts both has value, we take the latest timestamp
+   * @throws Exception
+   */
+  @Test
+  public void testDatasetDeletedTsMigration() throws Exception {
+    Account testAccount = makeTestAccountWithContainer();
+    Container testContainer = new ArrayList<>(testAccount.getAllContainers()).get(0);
+    String testDataset = DATASET_NAME_BASIC;
+
+    // add dataset with must provide info
+    Dataset dataset = new DatasetBuilder(testAccount.getName(), testContainer.getName(), testDataset).setVersionSchema(
+        Dataset.VersionSchema.MONOTONIC).build();
+    mySqlAccountStore.addDataset(testAccount.getId(), testContainer.getId(), dataset);
+
+    // delete a dataset, both deleted_ts and delete_ts are updated.
+    mySqlAccountStore.deleteDataset(testAccount.getId(), testContainer.getId(), testDataset);
+    // simulate the case, only delete_ts is update but deleted_ts is null
+    String query = "update Datasets set deleted_ts=NULL where accountId=" + testAccount.getId() + " and containerId="
+        + testContainer.getId() + " and datasetName=\"" + testDataset + "\"";
+    runSql(query);
+    // should return deleted even deleted_ts is NULL
+    try {
+      mySqlAccountStore.getDataset(testAccount.getId(), testContainer.getId(), testAccount.getName(),
+          testContainer.getName(), testDataset);
+      fail("Should fail due to the dataset deleted");
+    } catch (AccountServiceException e) {
+      assertEquals("Mistmatch on error code", AccountServiceErrorCode.Deleted, e.getErrorCode());
+    }
+
+    // list all datasets, expect return none
+    Page<String> allDatasets = mySqlAccountStore.listAllValidDatasets(testAccount.getId(), testContainer.getId(), null);
+    assertEquals("Mismatch for datasets list", 0, allDatasets.getEntries().size());
+
+    int retentionCount = 5;
+    // add new dataset with same primary key after it has been deleted.
+    dataset = new DatasetBuilder(testAccount.getName(), testContainer.getName(), testDataset).setVersionSchema(
+        Dataset.VersionSchema.TIMESTAMP).setRetentionTimeInSeconds(-1).setRetentionCount(retentionCount).build();
+    mySqlAccountStore.addDataset(testAccount.getId(), testContainer.getId(), dataset);
+
+    // delete a dataset
+    mySqlAccountStore.deleteDataset(testAccount.getId(), testContainer.getId(), testDataset);
+    // simulate the case, only delete_ts is expired but deleted_ts is not. Will pick the latest one.
+    query = "update Datasets set deleted_ts=DATE_ADD(now(), INTERVAL 1 DAY) where accountId=" + testAccount.getId()
+        + " and containerId=" + testContainer.getId() + " and datasetName=\"" + testDataset + "\"";
+    runSql(query);
+    // shouldn't return deleted_exception since one of the timestamp is not expired yet.
+    Dataset datasetFromMysql =
+        mySqlAccountStore.getDataset(testAccount.getId(), testContainer.getId(), testAccount.getName(),
+            testContainer.getName(), testDataset);
+    verifyDatasetProperties(dataset, datasetFromMysql);
+  }
+
   /**
    * Test add and get latest version.
    * @throws Exception
@@ -1421,9 +1476,20 @@ private void verifyDatasetProperties(Dataset expected, Dataset actual) {
     assertEquals("Mismatch on DatasetName of the dataset", expected.getDatasetName(), actual.getDatasetName());
     assertEquals("Mismatch on VersionSchema of the dataset", expected.getVersionSchema(), actual.getVersionSchema());
     assertEquals("Mismatch on RetentionCount of the dataset", expected.getRetentionCount(), actual.getRetentionCount());
-    assertEquals("Mismatch on RetentionTimeInSeconds of the dataset", expected.getRetentionTimeInSeconds(), actual.getRetentionTimeInSeconds());
+    assertEquals("Mismatch on RetentionTimeInSeconds of the dataset", expected.getRetentionTimeInSeconds(),
+        actual.getRetentionTimeInSeconds());
     assertEquals("Mismatch on UserTags of the dataset", expected.getUserTags(), actual.getUserTags());
+  }
 
+  private void runSql(String query) throws SQLException {
+    try {
+      try (Connection connection = mySqlAccountStore.getMySqlDataAccessor().getDatabaseConnection(true)) {
+        Statement statement = connection.createStatement();
+        statement.executeUpdate(query);
+      }
+    } catch (SQLException e) {
+      throw e;
+    }
   }
 
   /**

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

@@ -186,7 +186,7 @@ public DatasetDao(MySqlDataAccessor dataAccessor, MySqlAccountServiceConfig mySq
         DATASET_VERSION_STATE, CREATION_TIME, DELETED_TS);
     //this is to update the dataset version to permanent.
     updateDatasetVersionTtlSql = String.format(
-        "update %1$s set %2$s = NULL and %8$s = NULL where %3$s = ? and %4$s = ? and %5$s = ? and %6$s = ? and "
+        "update %1$s set %2$s = NULL, %8$s = NULL where %3$s = ? and %4$s = ? and %5$s = ? and %6$s = ? and "
             + "(COALESCE(%2$s, %8$s) is NULL or COALESCE(%2$s, %8$s) > now(3)) and %7$s = ?", DATASET_VERSION_TABLE,
         DELETE_TS, ACCOUNT_ID, CONTAINER_ID, DATASET_NAME, VERSION, DATASET_VERSION_STATE, DELETED_TS);
     //get the dataset version.

ambry-utils/src/main/java/com/github/ambry/utils/Utils.java

@@ -1503,7 +1503,10 @@ public static int compareTimestamp(Timestamp timestamp, long otherTimeMs) {
    * Get the latest timestamp, and if one of the timestamp is null, return the non-null timestamp.
    */
   public static Timestamp getLatestTimeStamp(Timestamp timestamp1, Timestamp timestamp2) {
-    if (compareTimestamp(timestamp1, timestampToMs(timestamp2)) > 0) {
+    if (timestamp1 == null || timestamp2 == null) {
+      return timestamp1 != null ? timestamp1 : timestamp2;
+    }
+    if (timestamp1.getTime() > timestamp2.getTime()) {
       return timestamp1;
     } else {
       return timestamp2;