Added delete container sql statement

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

@@ -154,7 +154,7 @@ public void testInitCacheOnStartUp() throws Exception {
     Account testAccount = makeTestAccountWithContainer();
     AccountUpdateInfo accountUpdateInfo =
         new AccountUpdateInfo(testAccount, true, false, new ArrayList<>(testAccount.getAllContainers()),
-            new ArrayList<>());
+            new ArrayList<>(), new ArrayList<>());
     mySqlAccountStore.updateAccounts(Collections.singletonList(accountUpdateInfo));
     mySqlAccountService = getAccountService();
     // Test in-memory cache is updated with accounts from mysql store on start up.
@@ -267,7 +267,7 @@ public void testBackgroundUpdater() throws Exception {
     // Add account to DB (could use second AS for this)
     mySqlAccountStore.updateAccounts(Collections.singletonList(
         new AccountUpdateInfo(testAccount, true, false, new ArrayList<>(testAccount.getAllContainers()),
-            new ArrayList<>())));
+            new ArrayList<>(), new ArrayList<>())));
 
     // sleep for polling interval time
     Thread.sleep(Long.parseLong(mySqlConfigProps.getProperty(UPDATER_POLLING_INTERVAL_SECONDS)) * 1000 + 100);

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

@@ -226,14 +226,16 @@ public static class AccountUpdateInfo {
     private final boolean isUpdated;
     private final List<Container> addedContainers;
     private final List<Container> updatedContainers;
+    private final List<Container> deletedContainers;
 
     public AccountUpdateInfo(Account account, boolean isAdded, boolean isUpdated, List<Container> addedContainers,
-        List<Container> updatedContainers) {
+        List<Container> updatedContainers, List<Container> deletedContainers) {
       this.account = account;
       this.isAdded = isAdded;
       this.isUpdated = isUpdated;
       this.addedContainers = addedContainers;
       this.updatedContainers = updatedContainers;
+      this.deletedContainers = deletedContainers;
     }
 
     /**
@@ -271,5 +273,12 @@ public List<Container> getAddedContainers() {
     public List<Container> getUpdatedContainers() {
       return updatedContainers;
     }
+
+    /**
+     * @return list of deleted {@link Container}s
+     */
+    public List<Container> getDeletedContainers() {
+      return deletedContainers;
+    }
   }
 }

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

@@ -763,7 +763,7 @@ protected void updateAccountsWithMySqlStore(Collection<Account> accounts) throws
       }
 
       accountsUpdateInfo.add(
-          new AccountUpdateInfo(account, isAccountAdded, isAccountUpdated, addedContainers, updatedContainers));
+          new AccountUpdateInfo(account, isAccountAdded, isAccountUpdated, addedContainers, updatedContainers, new ArrayList<>()));
     }
 
     // Write changes to MySql db.
@@ -788,7 +788,7 @@ private void updateContainersWithMySqlStore(Account account, Collection<Containe
     Pair<List<Container>, List<Container>> addedOrUpdatedContainers = getUpdatedContainers(account, containers);
     AccountUpdateInfo accountUpdateInfo =
         new AccountUpdateInfo(account, false, false, addedOrUpdatedContainers.getFirst(),
-            addedOrUpdatedContainers.getSecond());
+            addedOrUpdatedContainers.getSecond(), new ArrayList<>());
 
     // Write changes to MySql db.
     mySqlAccountStore.updateAccounts(Collections.singletonList(accountUpdateInfo));

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

@@ -68,6 +68,7 @@ public class AccountDao {
   // Container table query strings
   private final String insertContainersSql;
   private final String updateContainersSql;
+  private final String deleteContainersSql;
   private final String getContainersSinceSql;
   private final String getContainersByAccountSql;
   private final String getContainerByNameSql;
@@ -107,6 +108,8 @@ public AccountDao(MySqlDataAccessor dataAccessor) {
         String.format("update %s set %s = ?, %s = ?, %s = ?, %s = ?, %s = now(3) where %s = ? AND %s = ? ",
             CONTAINER_TABLE, CONTAINER_INFO, VERSION, CONTAINER_NAME, CONTAINER_STATUS, LAST_MODIFIED_TIME, ACCOUNT_ID,
             CONTAINER_ID);
+    deleteContainersSql =
+        String.format("delete from %s where %s = ? and %s = ?", CONTAINER_TABLE, ACCOUNT_ID, CONTAINER_ID);
     getContainerByNameSql =
         String.format("select %s, %s, %s, %s from %s where %s = ? and %s = ?", ACCOUNT_ID, CONTAINER_INFO, VERSION,
             LAST_MODIFIED_TIME, CONTAINER_TABLE, ACCOUNT_ID, CONTAINER_NAME);
@@ -292,7 +295,7 @@ private List<Container> convertContainersResultSet(ResultSet resultSet) throws S
   }
 
   /**
-   * Adds/Updates accounts and their containers to the database in batches atomically using transaction.
+   * Adds/Updates accounts and Add/updates/deletes their containers to the database in batches atomically using transaction.
    * @param accountsInfo information of updated Accounts
    * @param batchSize number of statements to be executed in one batch
    * @throws SQLException
@@ -305,7 +308,8 @@ public synchronized void updateAccounts(List<AccountUpdateInfo> accountsInfo, in
       accountUpdateBatch = new AccountUpdateBatch(dataAccessor.getPreparedStatement(insertAccountsSql, true),
           dataAccessor.getPreparedStatement(updateAccountsSql, true),
           dataAccessor.getPreparedStatement(insertContainersSql, true),
-          dataAccessor.getPreparedStatement(updateContainersSql, true));
+          dataAccessor.getPreparedStatement(updateContainersSql, true),
+          dataAccessor.getPreparedStatement(deleteContainersSql, true));
 
       // Disable auto commits
       dataAccessor.setAutoCommit(false);
@@ -318,10 +322,12 @@ public synchronized void updateAccounts(List<AccountUpdateInfo> accountsInfo, in
         boolean isAccountUpdated = accountUpdateInfo.isUpdated();
         List<Container> addedContainers = accountUpdateInfo.getAddedContainers();
         List<Container> updatedContainers = accountUpdateInfo.getUpdatedContainers();
+        List<Container> deletedContainers = accountUpdateInfo.getDeletedContainers();
 
         // Number of changes in the account.
         int accountUpdateCount =
-            (isAccountAdded ? 1 : 0) + (isAccountUpdated ? 1 : 0) + addedContainers.size() + updatedContainers.size();
+            (isAccountAdded ? 1 : 0) + (isAccountUpdated ? 1 : 0) + addedContainers.size() + updatedContainers.size()
+                + deletedContainers.size();
 
         // Commit transaction with previous batch inserts/updates if it either of following is true.
         // a) Total batch count of previous #accounts/containers is equal to or greater than configured batch size.
@@ -348,6 +354,10 @@ public synchronized void updateAccounts(List<AccountUpdateInfo> accountsInfo, in
         for (Container container : updatedContainers) {
           accountUpdateBatch.updateContainer(account.getId(), container);
         }
+        // Add deleted containers for batch deletes
+        for (Container container : deletedContainers) {
+          accountUpdateBatch.deleteContainer(account.getId(), container);
+        }
 
         batchCount += accountUpdateCount;
       }
@@ -435,6 +445,10 @@ private void bindContainer(PreparedStatement statement, int accountId, Container
         statement.setString(4, container.getStatus().toString());
         statement.setInt(5, accountId);
         statement.setInt(6, container.getId());
+        break;
+      case Delete:
+        statement.setInt(1, accountId);
+        statement.setInt(2, container.getId());
     }
   }
 
@@ -447,17 +461,21 @@ class AccountUpdateBatch {
     private int updateAccountCount = 0;
     private int insertContainerCount = 0;
     private int updateContainerCount = 0;
+    private int deleteContainerCount = 0;
     private final PreparedStatement insertAccountStatement;
     private final PreparedStatement updateAccountStatement;
     private final PreparedStatement insertContainerStatement;
     private final PreparedStatement updateContainerStatement;
+    private final PreparedStatement deleteContainerStatement;
 
     public AccountUpdateBatch(PreparedStatement insertAccountStatement, PreparedStatement updateAccountStatement,
-        PreparedStatement insertContainerStatement, PreparedStatement updateContainerStatement) {
+        PreparedStatement insertContainerStatement, PreparedStatement updateContainerStatement,
+        PreparedStatement deleteContainerStatement) {
       this.insertAccountStatement = insertAccountStatement;
       this.updateAccountStatement = updateAccountStatement;
       this.insertContainerStatement = insertContainerStatement;
       this.updateContainerStatement = updateContainerStatement;
+      this.deleteContainerStatement = deleteContainerStatement;
     }
 
     /**
@@ -481,6 +499,10 @@ public void maybeExecuteBatch() throws SQLException {
         updateContainerStatement.executeBatch();
         updateContainerCount = 0;
       }
+      if(deleteContainerStatement != null && deleteContainerCount > 0) {
+        deleteContainerStatement.executeBatch();
+        deleteContainerCount = 0;
+      }
     }
 
     /**
@@ -528,5 +550,17 @@ public void updateContainer(int accountId, Container container) throws SQLExcept
       updateContainerStatement.addBatch();
       ++updateContainerCount;
     }
+
+    /**
+     * Adds {@link Container} to its delete {@link PreparedStatement}'s batch.
+     * @param accountId account id of the Container.
+     * @param container {@link Container} to be deleted.
+     * @throws SQLException
+     */
+    public void deleteContainer(int accountId, Container container) throws SQLException {
+      bindContainer(deleteContainerStatement, accountId, container, StatementType.Delete);
+      deleteContainerStatement.addBatch();
+      ++deleteContainerCount;
+    }
   }
 }

ambry-account/src/test/java/com/github/ambry/account/mysql/AccountDaoTest.java

@@ -73,7 +73,7 @@ public AccountDaoTest() throws Exception {
             .lastModifiedTime(lastModifiedTime)
             .build();
     testAccountInfo = new AccountUpdateInfo(testAccount, true, false, new ArrayList<>(testAccount.getAllContainers()),
-        new ArrayList<>());
+        new ArrayList<>(), new ArrayList<>());
 
     Connection mockConnection = mock(Connection.class);
     MySqlDataAccessor dataAccessor = getDataAccessor(mockConnection, metrics);
@@ -211,7 +211,7 @@ public void testBatchOperations() throws SQLException {
     // test batch account inserts
     for (int i = 1; i <= size; i++) {
       Account account = new AccountBuilder((short) i, "test account " + i, Account.AccountStatus.ACTIVE).build();
-      accountUpdateInfos.add(new AccountUpdateInfo(account, true, false, new ArrayList<>(), new ArrayList<>()));
+      accountUpdateInfos.add(new AccountUpdateInfo(account, true, false, new ArrayList<>(), new ArrayList<>(), new ArrayList<>()));
     }
     accountDao.updateAccounts(accountUpdateInfos, batchSize);
     verify(mockAccountInsertStatement, times(size)).addBatch();
@@ -222,7 +222,7 @@ public void testBatchOperations() throws SQLException {
     for (int i = 1; i <= size; i++) {
       Account account =
           new AccountBuilder((short) i, "test account " + i, Account.AccountStatus.ACTIVE).snapshotVersion(1).build();
-      accountUpdateInfos.add(new AccountUpdateInfo(account, false, true, new ArrayList<>(), new ArrayList<>()));
+      accountUpdateInfos.add(new AccountUpdateInfo(account, false, true, new ArrayList<>(), new ArrayList<>(), new ArrayList<>()));
     }
     accountDao.updateAccounts(accountUpdateInfos, batchSize);
     verify(mockAccountUpdateStatement, times(size)).addBatch();
@@ -236,15 +236,15 @@ public void testBatchOperations() throws SQLException {
     }
     // test batch container inserts
     accountUpdateInfos.clear();
-    accountUpdateInfos.add(new AccountUpdateInfo(account, false, false, containers, new ArrayList<>()));
+    accountUpdateInfos.add(new AccountUpdateInfo(account, false, false, containers, new ArrayList<>(), new ArrayList<>()));
     accountDao.updateAccounts(accountUpdateInfos, batchSize);
     verify(mockContainerInsertStatement, times(size)).addBatch();
     // Execute batch should be invoked only once since all containers belong to same account
     verify(mockContainerInsertStatement, times(1)).executeBatch();
 
     // test batch container updates
     accountUpdateInfos.clear();
-    accountUpdateInfos.add(new AccountUpdateInfo(account, false, false, new ArrayList<>(), containers));
+    accountUpdateInfos.add(new AccountUpdateInfo(account, false, false, new ArrayList<>(), containers, new ArrayList<>()));
     accountDao.updateAccounts(accountUpdateInfos, batchSize);
     verify(mockContainerUpdateStatement, times(size)).addBatch();
     // Execute batch should be invoked only once since all containers belong to same account

ambry-tools/src/main/java/com/github/ambry/account/DatabaseTest.java

@@ -83,7 +83,7 @@ private static void perfTest(VerifiableProperties verifiableProperties) throws E
             .build());
         containersAdded++;
       }
-      accountUpdateInfos.add(new AccountUpdateInfo(account, true, false, containers, new ArrayList<>()));
+      accountUpdateInfos.add(new AccountUpdateInfo(account, true, false, containers, new ArrayList<>(), new ArrayList<>()));
     }
 
     accountDao.updateAccounts(accountUpdateInfos, 100);

ambry-tools/src/main/java/com/github/ambry/account/MySqlAccountsDBTool.java

@@ -227,7 +227,7 @@ public void initialize() throws SQLException {
     List<AccountUpdateInfo> accountUpdateInfos = new ArrayList<>();
     for (Account account : accountInfoMap.getAccounts()) {
       accountUpdateInfos.add(
-          new AccountUpdateInfo(account, true, false, new ArrayList<>(account.getAllContainers()), new ArrayList<>()));
+          new AccountUpdateInfo(account, true, false, new ArrayList<>(account.getAllContainers()), new ArrayList<>(), new ArrayList<>()));
     }
     mySqlAccountStore.updateAccounts(accountUpdateInfos);