ambari-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From stoa...@apache.org
Subject ambari git commit: AMBARI-20687. Perf: Refactor ambari db-cleanup to include all big tables.(vbrodetskyi)
Date Thu, 08 Jun 2017 11:51:00 GMT
Repository: ambari
Updated Branches:
  refs/heads/branch-2.5 24dcb1c85 -> c23ef5095


AMBARI-20687. Perf: Refactor ambari db-cleanup to include all big tables.(vbrodetskyi)


Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/c23ef509
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/c23ef509
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/c23ef509

Branch: refs/heads/branch-2.5
Commit: c23ef5095bbfffe910115b1cab4017f21f377e0a
Parents: 24dcb1c
Author: Vitaly Brodetskyi <vbrodetskyi@hortonworks.com>
Authored: Wed Apr 12 16:30:34 2017 +0300
Committer: Toader, Sebastian <stoader@hortonworks.com>
Committed: Thu Jun 8 11:51:51 2017 +0200

----------------------------------------------------------------------
 .../checks/DatabaseConsistencyCheckHelper.java  | 111 ++++++++
 .../apache/ambari/server/orm/DBAccessor.java    |   6 +
 .../ambari/server/orm/DBAccessorImpl.java       |   5 +
 .../server/orm/dao/HostRoleCommandDAO.java      |  16 ++
 .../ambari/server/orm/dao/RequestDAO.java       | 260 ++++++++++++++++++-
 .../server/orm/dao/TopologyHostTaskDAO.java     |  11 +
 .../orm/dao/TopologyLogicalRequestDAO.java      |  12 +
 .../server/orm/dao/TopologyLogicalTaskDAO.java  |  12 +
 .../orm/entities/ExecutionCommandEntity.java    |  15 +-
 .../orm/entities/HostRoleCommandEntity.java     |  10 +-
 .../server/orm/entities/RequestEntity.java      |   6 +
 .../entities/RequestOperationLevelEntity.java   |   4 +-
 .../entities/RequestResourceFilterEntity.java   |   5 +
 .../orm/entities/RoleSuccessCriteriaEntity.java |   5 +
 .../ambari/server/orm/entities/StageEntity.java |   6 +-
 .../orm/entities/TopologyHostRequestEntity.java |   5 +
 .../orm/entities/TopologyHostTaskEntity.java    |  15 +-
 .../entities/TopologyLogicalRequestEntity.java  |   5 +
 .../orm/entities/TopologyLogicalTaskEntity.java |  30 ++-
 .../server/orm/entities/UpgradeEntity.java      |   2 +
 .../server/orm/entities/UpgradeItemEntity.java  |   5 +
 ambari-server/src/main/python/ambari-server.py  |  10 +-
 .../src/main/python/ambari_server/dbCleanup.py  |  37 +--
 .../DatabaseConsistencyCheckHelperTest.java     |  66 +++++
 24 files changed, 623 insertions(+), 36 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelper.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelper.java b/ambari-server/src/main/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelper.java
index 4513fac..70c3661 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelper.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelper.java
@@ -24,6 +24,7 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -177,6 +178,7 @@ public class DatabaseConsistencyCheckHelper {
       checkHostComponentStatesCountEqualsHostComponentsDesiredStates();
       checkServiceConfigs();
       checkTopologyTables();
+      checkForLargeTables();
       LOG.info("******************************* Check database completed *******************************");
       return checkResult;
     }
@@ -269,6 +271,115 @@ public class DatabaseConsistencyCheckHelper {
   }
 
   /**
+   * This method checks if ambari database has tables with too big size (according to limit).
+   * First of all we are trying to get table size from schema information, but if it's not possible,
+   * we will get tables rows count and compare it with row count limit.
+   */
+  static void checkForLargeTables() {
+    LOG.info("Checking for tables with large physical size");
+
+    ensureConnection();
+
+    DBAccessor.DbType dbType = dbAccessor.getDbType();
+    String schemaName = dbAccessor.getDbSchema();
+
+    String GET_TABLE_SIZE_IN_BYTES_POSTGRESQL = "SELECT pg_total_relation_size('%s') \"Table Size\"";
+    String GET_TABLE_SIZE_IN_BYTES_MYSQL = "SELECT (data_length + index_length) \"Table Size\" FROM information_schema.TABLES WHERE table_schema = \"" + schemaName + "\" AND table_name =\"%s\"";
+    String GET_TABLE_SIZE_IN_BYTES_ORACLE = "SELECT bytes \"Table Size\" FROM user_segments WHERE segment_type='TABLE' AND segment_name='%s'";
+    String GET_ROW_COUNT_QUERY = "SELECT COUNT(*) FROM %s";
+
+    Map<DBAccessor.DbType, String> tableSizeQueryMap = new HashMap<>();
+    tableSizeQueryMap.put(DBAccessor.DbType.POSTGRES, GET_TABLE_SIZE_IN_BYTES_POSTGRESQL);
+    tableSizeQueryMap.put(DBAccessor.DbType.MYSQL, GET_TABLE_SIZE_IN_BYTES_MYSQL);
+    tableSizeQueryMap.put(DBAccessor.DbType.ORACLE, GET_TABLE_SIZE_IN_BYTES_ORACLE);
+
+    List<String> tablesToCheck = Arrays.asList("host_role_command", "execution_command", "stage", "request", "alert_history");
+
+    final double TABLE_SIZE_LIMIT_MB = 3000.0;
+    final int TABLE_ROW_COUNT_LIMIT = 3000000;
+
+    String findTableSizeQuery = tableSizeQueryMap.get(dbType);
+
+    if (dbType == DBAccessor.DbType.ORACLE) {
+      for (int i = 0;i < tablesToCheck.size(); i++) {
+        tablesToCheck.set(i, tablesToCheck.get(i).toUpperCase());
+      }
+    }
+
+    for (String tableName : tablesToCheck) {
+
+      ResultSet rs = null;
+      Statement statement = null;
+      Double tableSizeInMB = null;
+      Long tableSizeInBytes = null;
+      int tableRowCount = -1;
+
+      try {
+        statement = connection.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE);
+        rs = statement.executeQuery(String.format(findTableSizeQuery, tableName));
+        if (rs != null) {
+          while (rs.next()) {
+            tableSizeInBytes = rs.getLong(1);
+            if (tableSizeInBytes != null) {
+              tableSizeInMB = tableSizeInBytes / 1024.0 / 1024.0;
+            }
+          }
+        }
+
+        if (tableSizeInMB != null && tableSizeInMB > TABLE_SIZE_LIMIT_MB) {
+          warning("The database table {} is currently {} MB (limit is {}) and may impact performance. It is recommended " +
+                  "that you reduce its size by executing \"ambari-server db-cleanup\".",
+                  tableName, tableSizeInMB, TABLE_SIZE_LIMIT_MB);
+        } else if (tableSizeInMB != null && tableSizeInMB < TABLE_SIZE_LIMIT_MB) {
+          LOG.info(String.format("The database table %s is currently %.3f MB and is within normal limits (%.3f)",
+                  tableName, tableSizeInMB, TABLE_SIZE_LIMIT_MB));
+        } else {
+          throw new Exception();
+        }
+      } catch (Exception e) {
+        LOG.error(String.format("Failed to get %s table size from database, will check row count: ", tableName), e);
+        try {
+          rs = statement.executeQuery(String.format(GET_ROW_COUNT_QUERY, tableName));
+          if (rs != null) {
+            while (rs.next()) {
+              tableRowCount = rs.getInt(1);
+            }
+          }
+
+          if (tableRowCount > TABLE_ROW_COUNT_LIMIT) {
+            warning("The database table {} currently has {} rows (limit is {}) and may impact performance. It is " +
+                    "recommended that you reduce its size by executing \"ambari-server db-cleanup\".",
+                    tableName, tableRowCount, TABLE_ROW_COUNT_LIMIT);
+          } else if (tableRowCount != -1 && tableRowCount < TABLE_ROW_COUNT_LIMIT) {
+            LOG.info(String.format("The database table %s currently has %d rows and is within normal limits (%d)", tableName, tableRowCount, TABLE_ROW_COUNT_LIMIT));
+          } else {
+            throw new SQLException();
+          }
+        } catch (SQLException ex) {
+          LOG.error(String.format("Failed to get %s row count: ", tableName), e);
+        }
+      } finally {
+        if (rs != null) {
+          try {
+            rs.close();
+          } catch (SQLException e) {
+            LOG.error("Exception occurred during result set closing procedure: ", e);
+          }
+        }
+
+        if (statement != null) {
+          try {
+            statement.close();
+          } catch (SQLException e) {
+            LOG.error("Exception occurred during statement closing procedure: ", e);
+          }
+        }
+      }
+    }
+
+  }
+
+  /**
   * This method checks if any config type in clusterconfigmapping table, has
   * more than one versions selected. If config version is selected(in selected column = 1),
   * it means that this version of config is actual. So, if any config type has more

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java
index fac524c..c637c05 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java
@@ -663,6 +663,12 @@ public interface DBAccessor {
   DbType getDbType();
 
   /**
+   * Get database schema name
+   * @return @dbSchema
+   */
+  String getDbSchema();
+
+  /**
    * Capture column type
    */
   class DBColumnInfo {

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java
index c5b116c..0e2237c 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java
@@ -233,6 +233,11 @@ public class DBAccessorImpl implements DBAccessor {
   }
 
   @Override
+  public String getDbSchema() {
+    return dbSchema;
+  }
+
+  @Override
   public boolean tableHasData(String tableName) throws SQLException {
     String query = "SELECT count(*) from " + tableName;
     Statement statement = getConnection().createStatement();

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java
index 7582957..b9e1fab 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java
@@ -932,4 +932,20 @@ public class HostRoleCommandDAO {
       return HostRoleCommandEntity_.getPredicateMapping().get(propertyId);
     }
   }
+
+  public List<Long> findTaskIdsByRequestStageIds(List<RequestDAO.StageEntityPK> requestStageIds) {
+    EntityManager entityManager = entityManagerProvider.get();
+    List<Long> taskIds = new ArrayList<Long>();
+    for (RequestDAO.StageEntityPK requestIds : requestStageIds) {
+      TypedQuery<Long> hostRoleCommandQuery =
+              entityManager.createNamedQuery("HostRoleCommandEntity.findTaskIdsByRequestStageIds", Long.class);
+
+      hostRoleCommandQuery.setParameter("requestId", requestIds.getRequestId());
+      hostRoleCommandQuery.setParameter("stageId", requestIds.getStageId());
+
+      taskIds.addAll(daoUtils.selectList(hostRoleCommandQuery));
+    }
+
+    return taskIds;
+  }
 }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
index 1c4d0a3..38c0977 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestDAO.java
@@ -19,27 +19,54 @@
 package org.apache.ambari.server.orm.dao;
 
 import java.text.MessageFormat;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Set;
 
 import javax.persistence.EntityManager;
 import javax.persistence.TypedQuery;
 
+import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.cleanup.TimeBasedCleanupPolicy;
 import org.apache.ambari.server.orm.RequiresSession;
+import org.apache.ambari.server.orm.entities.ExecutionCommandEntity;
+import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
 import org.apache.ambari.server.orm.entities.RequestEntity;
+import org.apache.ambari.server.orm.entities.RequestOperationLevelEntity;
 import org.apache.ambari.server.orm.entities.RequestResourceFilterEntity;
+import org.apache.ambari.server.orm.entities.RoleSuccessCriteriaEntity;
+import org.apache.ambari.server.orm.entities.StageEntity;
+import org.apache.ambari.server.orm.entities.TopologyHostRequestEntity;
+import org.apache.ambari.server.orm.entities.TopologyHostTaskEntity;
+import org.apache.ambari.server.orm.entities.TopologyLogicalTaskEntity;
+import org.apache.ambari.server.state.Clusters;
 import org.eclipse.persistence.config.HintValues;
 import org.eclipse.persistence.config.QueryHints;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import com.google.inject.Singleton;
 import com.google.inject.persist.Transactional;
 
 @Singleton
-public class RequestDAO {
+public class RequestDAO implements Cleanable {
+
+  private static final Logger LOG = LoggerFactory.getLogger(RequestDAO.class);
+
+
+  private static final int BATCH_SIZE = 999;
+
   /**
    * SQL template to retrieve all request IDs, sorted by the ID.
    */
@@ -64,6 +91,27 @@ public class RequestDAO {
   @Inject
   DaoUtils daoUtils;
 
+  @Inject
+  private Provider<Clusters> m_clusters;
+
+  @Inject
+  private HostRoleCommandDAO hostRoleCommandDAO;
+
+  @Inject
+  private StageDAO stageDAO;
+
+  @Inject
+  private TopologyLogicalTaskDAO topologyLogicalTaskDAO;
+
+  @Inject
+  private TopologyHostTaskDAO topologyHostTaskDAO;
+
+  @Inject
+  private TopologyLogicalRequestDAO topologyLogicalRequestDAO;
+
+  @Inject
+  private TopologyRequestDAO topologyRequestDAO;
+
   @RequiresSession
   public RequestEntity findByPK(Long requestId) {
     return entityManagerProvider.get().find(RequestEntity.class, requestId);
@@ -189,4 +237,214 @@ public class RequestDAO {
 
     return daoUtils.selectList(query);
   }
+
+  public static final class StageEntityPK {
+    private Long requestId;
+    private Long stageId;
+
+    public StageEntityPK(Long requestId, Long stageId) {
+      this.requestId = requestId;
+      this.stageId = stageId;
+    }
+
+    public Long getStageId() {
+      return stageId;
+    }
+
+    public void setStageId(Long stageId) {
+      this.stageId = stageId;
+    }
+
+    public Long getRequestId() {
+      return requestId;
+    }
+
+    public void setRequestId(Long requestId) {
+      this.requestId = requestId;
+    }
+  }
+
+  /**
+   * Search for all request ids in Upgrade table
+   * @return the list of request ids
+   */
+  private List<Long> findAllRequestIdsFromUpgrade() {
+    EntityManager entityManager = entityManagerProvider.get();
+    TypedQuery<Long> upgradeQuery =
+            entityManager.createNamedQuery("UpgradeEntity.findAllRequestIds", Long.class);
+
+    return daoUtils.selectList(upgradeQuery);
+  }
+
+  /**
+   * Search for all request and stage ids in Request and Stage tables
+   * @return the list of request/stage ids
+   */
+  public List<StageEntityPK> findRequestAndStageIdsInClusterBeforeDate(Long clusterId, long  beforeDateMillis) {
+    EntityManager entityManager = entityManagerProvider.get();
+    TypedQuery<StageEntityPK> requestQuery =
+            entityManager.createNamedQuery("RequestEntity.findRequestStageIdsInClusterBeforeDate", StageEntityPK.class);
+
+    requestQuery.setParameter("clusterId", clusterId);
+    requestQuery.setParameter("beforeDate", beforeDateMillis);
+
+    return daoUtils.selectList(requestQuery);
+  }
+
+  /**
+   * In this method we are removing entities using passed ids,
+   * To prevent issues we are using batch request to remove limited
+   * count of entities.
+   * @param ids              list of ids that we are using to remove rows from table
+   * @param paramName        name of parameter that we are using in sql query (taskIds, stageIds)
+   * @param entityName       name of entity which we will remove
+   * @param beforeDateMillis timestamp which was set by user (remove all entities that were created before),
+   *                         we are using it only for logging
+   * @param entityQuery      name of NamedQuery which we will use to remove needed entities
+   * @param type             type of entity class which we will use for casting query result
+   * @return                 rows count that were removed
+   */
+  @Transactional
+  protected <T> int cleanTableByIds(Set<Long> ids, String paramName, String entityName, Long beforeDateMillis,
+                                  String entityQuery, Class<T> type) {
+    LOG.info(String.format("Deleting %s entities before date %s", entityName, new Date(beforeDateMillis)));
+    EntityManager entityManager = entityManagerProvider.get();
+    int affectedRows = 0;
+    // Batch delete
+    TypedQuery<T> query = entityManager.createNamedQuery(entityQuery, type);
+    if (ids != null && !ids.isEmpty()) {
+      for (int i = 0; i < ids.size(); i += BATCH_SIZE) {
+        int endRow = (i + BATCH_SIZE) > ids.size() ? ids.size() : (i + BATCH_SIZE);
+        List<Long> idsSubList = new ArrayList<>(ids).subList(i, endRow);
+        LOG.info("Deleting " + entityName + " entity batch with task ids: " +
+                idsSubList.get(0) + " - " + idsSubList.get(idsSubList.size() - 1));
+        query.setParameter(paramName, idsSubList);
+        affectedRows += query.executeUpdate();
+      }
+    }
+
+    return affectedRows;
+  }
+
+  /**
+   * In this method we are removing entities using passed few ids,
+   * To prevent issues we are using batch request to remove limited
+   * count of entities.
+   * @param ids              list of ids pairs that we are using to remove rows from table
+   * @param paramNames       list of two names of parameters that we are using in sql query (taskIds, stageIds)
+   * @param entityName       name of entity which we will remove
+   * @param beforeDateMillis timestamp which was set by user (remove all entities that were created before),
+   *                         we are using it only for logging
+   * @param entityQuery      name of NamedQuery which we will use to remove needed entities
+   * @param type             type of entity class which we will use for casting query result
+   * @return                 rows count that were removed
+   */
+  @Transactional
+  protected <T> int cleanTableByStageEntityPK(List<StageEntityPK> ids, LinkedList<String> paramNames, String entityName, Long beforeDateMillis,
+                                  String entityQuery, Class<T> type) {
+    LOG.info(String.format("Deleting %s entities before date %s", entityName, new Date(beforeDateMillis)));
+    EntityManager entityManager = entityManagerProvider.get();
+    int affectedRows = 0;
+    // Batch delete
+    TypedQuery<T> query = entityManager.createNamedQuery(entityQuery, type);
+    if (ids != null && !ids.isEmpty()) {
+      for (int i = 0; i < ids.size(); i += BATCH_SIZE) {
+        int endRow = (i + BATCH_SIZE) > ids.size() ? ids.size() : (i + BATCH_SIZE);
+        List<StageEntityPK> idsSubList = new ArrayList<>(ids).subList(i, endRow);
+        LOG.info("Deleting " + entityName + " entity batch with task ids: " +
+                idsSubList.get(0) + " - " + idsSubList.get(idsSubList.size() - 1));
+        for (StageEntityPK requestIds : idsSubList) {
+          query.setParameter(paramNames.get(0), requestIds.getStageId());
+          query.setParameter(paramNames.get(1), requestIds.getRequestId());
+          affectedRows += query.executeUpdate();
+        }
+      }
+    }
+
+    return affectedRows;
+  }
+
+  @Transactional
+  @Override
+  public long cleanup(TimeBasedCleanupPolicy policy) {
+    long affectedRows = 0;
+    Long clusterId = null;
+    try {
+      clusterId = m_clusters.get().getCluster(policy.getClusterName()).getClusterId();
+      // find request and stage ids that were created before date populated by user.
+      List<StageEntityPK> requestStageIds = findRequestAndStageIdsInClusterBeforeDate(clusterId, policy.getToDateInMillis());
+
+      // find request ids from Upgrade table and exclude these ids from
+      // request ids set that we already have. We don't want to make any changes for upgrade
+      Set<Long> requestIdsFromUpgrade = Sets.newHashSet(findAllRequestIdsFromUpgrade());
+      Iterator<StageEntityPK> requestStageIdsIterator =  requestStageIds.iterator();
+      while (requestStageIdsIterator.hasNext()) {
+        StageEntityPK nextRequestStageIds = requestStageIdsIterator.next();
+        if (requestIdsFromUpgrade.contains(nextRequestStageIds.getRequestId())) {
+          requestStageIdsIterator.remove();
+        }
+      }
+
+
+      Set<Long> requestIds = new HashSet<>();
+      for (StageEntityPK ids : requestStageIds) {
+        requestIds.add(ids.getRequestId());
+      }
+
+      // find task ids using request stage ids
+      Set<Long> taskIds = Sets.newHashSet(hostRoleCommandDAO.findTaskIdsByRequestStageIds(requestStageIds));
+      LinkedList<String> params = new LinkedList<>();
+      params.add("stageId");
+      params.add("requestId");
+
+      // find host task ids, to find related host requests and also to remove needed host tasks
+      List<Long> hostTaskIds = new ArrayList<>();
+      if (taskIds != null && !taskIds.isEmpty()) {
+        hostTaskIds = topologyLogicalTaskDAO.findHostTaskIdsByPhysicalTaskIds(Lists.newArrayList(taskIds));
+      }
+
+      // find host request ids by host task ids to remove later needed host requests
+      List<Long> hostRequestIds = new ArrayList<>();
+      if (!hostTaskIds.isEmpty()) {
+        hostRequestIds = topologyHostTaskDAO.findHostRequestIdsByHostTaskIds(hostTaskIds);
+      }
+
+      List<Long> topologyRequestIds = new ArrayList<>();
+      if (!hostRequestIds.isEmpty()) {
+        topologyRequestIds = topologyLogicalRequestDAO.findRequestIdsByIds(hostRequestIds);
+      }
+
+
+      //removing all entities one by one according to their relations using stage, task and request ids
+      affectedRows += cleanTableByIds(taskIds, "taskIds", "ExecutionCommand", policy.getToDateInMillis(),
+              "ExecutionCommandEntity.removeByTaskIds", ExecutionCommandEntity.class);
+      affectedRows += cleanTableByIds(taskIds, "taskIds", "TopologyLogicalTask", policy.getToDateInMillis(),
+              "TopologyLogicalTaskEntity.removeByPhysicalTaskIds", TopologyLogicalTaskEntity.class);
+      affectedRows += cleanTableByIds(Sets.newHashSet(hostTaskIds), "hostTaskIds", "TopologyHostTask", policy.getToDateInMillis(),
+              "TopologyHostTaskEntity.removeByTaskIds", TopologyHostTaskEntity.class);
+      affectedRows += cleanTableByIds(Sets.newHashSet(hostRequestIds), "hostRequestIds", "TopologyHostRequest", policy.getToDateInMillis(),
+              "TopologyHostRequestEntity.removeByIds", TopologyHostRequestEntity.class);
+      for (Long topologyRequestId : topologyRequestIds) {
+        topologyRequestDAO.removeByPK(topologyRequestId);
+      }
+      affectedRows += cleanTableByIds(taskIds, "taskIds", "HostRoleCommand", policy.getToDateInMillis(),
+              "HostRoleCommandEntity.removeByTaskIds", HostRoleCommandEntity.class);
+      affectedRows += cleanTableByStageEntityPK(requestStageIds, params, "RoleSuccessCriteria", policy.getToDateInMillis(),
+              "RoleSuccessCriteriaEntity.removeByRequestStageIds", RoleSuccessCriteriaEntity.class);
+      affectedRows += cleanTableByStageEntityPK(requestStageIds, params, "Stage", policy.getToDateInMillis(),
+              "StageEntity.removeByRequestStageIds", StageEntity.class);
+      affectedRows += cleanTableByIds(requestIds, "requestIds", "RequestResourceFilter", policy.getToDateInMillis(),
+              "RequestResourceFilterEntity.removeByRequestIds", RequestResourceFilterEntity.class);
+      affectedRows += cleanTableByIds(requestIds, "requestIds", "RequestOperationLevel", policy.getToDateInMillis(),
+              "RequestOperationLevelEntity.removeByRequestIds", RequestOperationLevelEntity.class);
+      affectedRows += cleanTableByIds(requestIds, "requestIds", "Request", policy.getToDateInMillis(),
+              "RequestEntity.removeByRequestIds", RequestEntity.class);
+
+    } catch (AmbariException e) {
+      LOG.error("Error while looking up cluster with name: {}", policy.getClusterName(), e);
+      throw new IllegalStateException(e);
+    }
+
+    return affectedRows;
+  }
 }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyHostTaskDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyHostTaskDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyHostTaskDAO.java
index 85a4f5f..eea8032 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyHostTaskDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyHostTaskDAO.java
@@ -52,6 +52,17 @@ public class TopologyHostTaskDAO {
   }
 
   @RequiresSession
+  public List<Long> findHostRequestIdsByHostTaskIds(List<Long> hostTaskIds) {
+    EntityManager entityManager = entityManagerProvider.get();
+    TypedQuery<Long> topologyHostTaskQuery =
+            entityManager.createNamedQuery("TopologyLogicalTaskEntity.findHostRequestIdsByHostTaskIds", Long.class);
+
+    topologyHostTaskQuery.setParameter("hostTaskIds", hostTaskIds);
+
+    return daoUtils.selectList(topologyHostTaskQuery);
+  }
+
+  @RequiresSession
   public List<TopologyHostTaskEntity> findAll() {
     return daoUtils.selectAll(entityManagerProvider.get(), TopologyHostTaskEntity.class);
   }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalRequestDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalRequestDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalRequestDAO.java
index e6dcb69..32a38da 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalRequestDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalRequestDAO.java
@@ -24,6 +24,7 @@ import com.google.inject.persist.Transactional;
 import org.apache.ambari.server.orm.RequiresSession;
 import org.apache.ambari.server.orm.entities.TopologyLogicalRequestEntity;
 import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
 import java.util.List;
 
 @Singleton
@@ -58,4 +59,15 @@ public class TopologyLogicalRequestDAO {
   public void remove(TopologyLogicalRequestEntity requestEntity) {
     entityManagerProvider.get().remove(requestEntity);
   }
+
+  @RequiresSession
+  public List<Long> findRequestIdsByIds(List<Long> ids) {
+    EntityManager entityManager = entityManagerProvider.get();
+    TypedQuery<Long> topologyLogicalRequestQuery =
+            entityManager.createNamedQuery("TopologyLogicalRequestEntity.findRequestIds", Long.class);
+
+    topologyLogicalRequestQuery.setParameter("ids", ids);
+
+    return daoUtils.selectList(topologyLogicalRequestQuery);
+  }
 }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalTaskDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalTaskDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalTaskDAO.java
index f0331cc..3a72aed 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalTaskDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/TopologyLogicalTaskDAO.java
@@ -25,6 +25,7 @@ import org.apache.ambari.server.orm.RequiresSession;
 import org.apache.ambari.server.orm.entities.TopologyLogicalTaskEntity;
 
 import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
 import java.util.List;
 
 @Singleton
@@ -41,6 +42,17 @@ public class TopologyLogicalTaskDAO {
   }
 
   @RequiresSession
+  public List<Long> findHostTaskIdsByPhysicalTaskIds(List<Long> physicalTaskIds) {
+    EntityManager entityManager = entityManagerProvider.get();
+    TypedQuery<Long> topologyHostTaskQuery =
+            entityManager.createNamedQuery("TopologyLogicalTaskEntity.findHostTaskIdsByPhysicalTaskIds", Long.class);
+
+    topologyHostTaskQuery.setParameter("physicalTaskIds", physicalTaskIds);
+
+    return daoUtils.selectList(topologyHostTaskQuery);
+  }
+
+  @RequiresSession
   public List<TopologyLogicalTaskEntity> findAll() {
     return daoUtils.selectAll(entityManagerProvider.get(), TopologyLogicalTaskEntity.class);
   }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ExecutionCommandEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ExecutionCommandEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ExecutionCommandEntity.java
index 25d830b..7015709 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ExecutionCommandEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ExecutionCommandEntity.java
@@ -18,11 +18,24 @@
 
 package org.apache.ambari.server.orm.entities;
 
-import javax.persistence.*;
 import java.util.Arrays;
 
+import javax.persistence.Basic;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.JoinColumn;
+import javax.persistence.Lob;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
+import javax.persistence.OneToOne;
+import javax.persistence.Table;
+
 @Table(name = "execution_command")
 @Entity
+@NamedQueries({
+    @NamedQuery(name = "ExecutionCommandEntity.removeByTaskIds", query = "DELETE FROM ExecutionCommandEntity command WHERE command.taskId IN :taskIds")
+})
 public class ExecutionCommandEntity {
 
   @Id

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostRoleCommandEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostRoleCommandEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostRoleCommandEntity.java
index 3d946f5..7ac60a9 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostRoleCommandEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostRoleCommandEntity.java
@@ -60,6 +60,7 @@ import org.apache.commons.lang.ArrayUtils;
     , initialValue = 1
 )
 @NamedQueries({
+    @NamedQuery(name = "HostRoleCommandEntity.findTaskIdsByRequestStageIds", query = "SELECT command.taskId FROM HostRoleCommandEntity command WHERE command.stageId = :stageId AND command.requestId = :requestId"),
     @NamedQuery(name = "HostRoleCommandEntity.findCountByCommandStatuses", query = "SELECT COUNT(command.taskId) FROM HostRoleCommandEntity command WHERE command.status IN :statuses"),
     @NamedQuery(name = "HostRoleCommandEntity.findByRequestIdAndStatuses", query="SELECT task FROM HostRoleCommandEntity task WHERE task.requestId=:requestId AND task.status IN :statuses ORDER BY task.taskId ASC"),
     @NamedQuery(name = "HostRoleCommandEntity.findTasksByStatusesOrderByIdDesc", query = "SELECT task FROM HostRoleCommandEntity task WHERE task.requestId = :requestId AND task.status IN :statuses ORDER BY task.taskId DESC"),
@@ -71,12 +72,9 @@ import org.apache.commons.lang.ArrayUtils;
     @NamedQuery(name = "HostRoleCommandEntity.findByStatusBetweenStages", query = "SELECT command FROM HostRoleCommandEntity command WHERE command.requestId = :requestId AND command.stageId >= :minStageId AND command.stageId <= :maxStageId AND command.status = :status"),
     @NamedQuery(name = "HostRoleCommandEntity.updateAutoSkipExcludeRoleCommand", query = "UPDATE HostRoleCommandEntity command SET command.autoSkipOnFailure = :autoSkipOnFailure WHERE command.requestId = :requestId AND command.roleCommand <> :roleCommand"),
     @NamedQuery(name = "HostRoleCommandEntity.updateAutoSkipForRoleCommand", query = "UPDATE HostRoleCommandEntity command SET command.autoSkipOnFailure = :autoSkipOnFailure WHERE command.requestId = :requestId AND command.roleCommand = :roleCommand"),
-    @NamedQuery(
-        name = "HostRoleCommandEntity.findHostsByCommandStatus",
-        query = "SELECT DISTINCT(host.hostName) FROM HostRoleCommandEntity command, HostEntity host WHERE (command.requestId >= :iLowestRequestIdInProgress AND command.requestId <= :iHighestRequestIdInProgress) AND command.status IN :statuses AND command.hostId = host.hostId AND host.hostName IS NOT NULL"),
-    @NamedQuery(
-        name = "HostRoleCommandEntity.getBlockingHostsForRequest",
-        query = "SELECT DISTINCT(host.hostName) FROM HostRoleCommandEntity command, HostEntity host WHERE command.requestId >= :lowerRequestIdInclusive AND command.requestId < :upperRequestIdExclusive AND command.status IN :statuses AND command.isBackgroundCommand=0 AND command.hostId = host.hostId AND host.hostName IS NOT NULL")
+    @NamedQuery(name = "HostRoleCommandEntity.removeByTaskIds", query = "DELETE FROM HostRoleCommandEntity command WHERE command.taskId IN :taskIds"),
+    @NamedQuery(name = "HostRoleCommandEntity.findHostsByCommandStatus", query = "SELECT DISTINCT(host.hostName) FROM HostRoleCommandEntity command, HostEntity host WHERE (command.requestId >= :iLowestRequestIdInProgress AND command.requestId <= :iHighestRequestIdInProgress) AND command.status IN :statuses AND command.hostId = host.hostId AND host.hostName IS NOT NULL"),
+    @NamedQuery(name = "HostRoleCommandEntity.getBlockingHostsForRequest", query = "SELECT DISTINCT(host.hostName) FROM HostRoleCommandEntity command, HostEntity host WHERE command.requestId >= :lowerRequestIdInclusive AND command.requestId < :upperRequestIdExclusive AND command.status IN :statuses AND command.isBackgroundCommand=0 AND command.hostId = host.hostId AND host.hostName IS NOT NULL")
 
 })
 public class HostRoleCommandEntity {

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestEntity.java
index e46bb51..45fb631 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestEntity.java
@@ -32,6 +32,8 @@ import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.Lob;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.OneToMany;
 import javax.persistence.OneToOne;
 import javax.persistence.Table;
@@ -39,6 +41,10 @@ import java.util.Collection;
 
 @Table(name = "request")
 @Entity
+@NamedQueries({
+  @NamedQuery(name = "RequestEntity.findRequestStageIdsInClusterBeforeDate", query = "SELECT NEW org.apache.ambari.server.orm.dao.RequestDAO.StageEntityPK(request.requestId, stage.stageId) FROM RequestEntity request JOIN StageEntity stage ON request.requestId = stage.requestId WHERE request.clusterId = :clusterId AND request.createTime <= :beforeDate"),
+  @NamedQuery(name = "RequestEntity.removeByRequestIds", query = "DELETE FROM RequestEntity request WHERE request.requestId IN :requestIds")
+})
 public class RequestEntity {
 
   @Column(name = "request_id")

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestOperationLevelEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestOperationLevelEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestOperationLevelEntity.java
index c03816e..64af92a 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestOperationLevelEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestOperationLevelEntity.java
@@ -42,7 +42,9 @@ import javax.persistence.TableGenerator;
 @NamedQueries({
     @NamedQuery(name = "requestOperationLevelByHostId", query =
         "SELECT requestOperationLevel FROM RequestOperationLevelEntity requestOperationLevel " +
-            "WHERE requestOperationLevel.hostId=:hostId")
+            "WHERE requestOperationLevel.hostId=:hostId"),
+    @NamedQuery(name = "RequestOperationLevelEntity.removeByRequestIds",
+        query = "DELETE FROM RequestOperationLevelEntity requestOperationLevel WHERE requestOperationLevel.requestId IN :requestIds")
 })
 public class RequestOperationLevelEntity {
 

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestResourceFilterEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestResourceFilterEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestResourceFilterEntity.java
index 8ee41d2..9597db1 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestResourceFilterEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestResourceFilterEntity.java
@@ -26,6 +26,8 @@ import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.Lob;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.Table;
 import javax.persistence.TableGenerator;
 
@@ -36,6 +38,9 @@ import javax.persistence.TableGenerator;
   , pkColumnValue = "resourcefilter_id_seq"
   , initialValue = 1
 )
+@NamedQueries({
+  @NamedQuery(name = "RequestResourceFilterEntity.removeByRequestIds", query = "DELETE FROM RequestResourceFilterEntity filter WHERE filter.requestId IN :requestIds")
+})
 public class RequestResourceFilterEntity {
 
   @Id

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RoleSuccessCriteriaEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RoleSuccessCriteriaEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RoleSuccessCriteriaEntity.java
index 3386c24..66e7fd8 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RoleSuccessCriteriaEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RoleSuccessCriteriaEntity.java
@@ -26,6 +26,8 @@ import javax.persistence.IdClass;
 import javax.persistence.JoinColumn;
 import javax.persistence.JoinColumns;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.Table;
 
 import org.apache.ambari.server.Role;
@@ -33,6 +35,9 @@ import org.apache.ambari.server.Role;
 @IdClass(org.apache.ambari.server.orm.entities.RoleSuccessCriteriaEntityPK.class)
 @Table(name = "role_success_criteria")
 @Entity
+@NamedQueries({
+  @NamedQuery(name = "RoleSuccessCriteriaEntity.removeByRequestStageIds", query = "DELETE FROM RoleSuccessCriteriaEntity criteria WHERE criteria.stageId = :stageId AND criteria.requestId = :requestId")
+})
 public class RoleSuccessCriteriaEntity {
 
   @Id

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity.java
index aeafda0..3b755f6 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity.java
@@ -49,7 +49,11 @@ import org.apache.ambari.server.actionmanager.CommandExecutionType;
         query = "SELECT stage.requestId, MIN(stage.stageId) from StageEntity stage, HostRoleCommandEntity hrc WHERE hrc.status IN :statuses AND hrc.stageId = stage.stageId AND hrc.requestId = stage.requestId GROUP by stage.requestId ORDER BY stage.requestId"),
     @NamedQuery(
         name = "StageEntity.findByRequestIdAndCommandStatuses",
-        query = "SELECT stage from StageEntity stage WHERE stage.stageId IN (SELECT roleCommand.stageId from HostRoleCommandEntity roleCommand WHERE roleCommand.requestId = :requestId AND roleCommand.status IN :statuses AND roleCommand.stageId = stage.stageId AND roleCommand.requestId = stage.requestId ) ORDER BY stage.stageId") })
+        query = "SELECT stage from StageEntity stage WHERE stage.stageId IN (SELECT roleCommand.stageId from HostRoleCommandEntity roleCommand WHERE roleCommand.requestId = :requestId AND roleCommand.status IN :statuses AND roleCommand.stageId = stage.stageId AND roleCommand.requestId = stage.requestId ) ORDER BY stage.stageId"),
+    @NamedQuery(
+        name = "StageEntity.removeByRequestStageIds",
+        query = "DELETE FROM StageEntity stage WHERE stage.stageId = :stageId AND stage.requestId = :requestId")
+})
 public class StageEntity {
 
   @Basic

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostRequestEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostRequestEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostRequestEntity.java
index 4e05ea1..7abbd51 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostRequestEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostRequestEntity.java
@@ -25,6 +25,8 @@ import javax.persistence.GenerationType;
 import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.OneToMany;
 import javax.persistence.Table;
 import javax.persistence.TableGenerator;
@@ -32,6 +34,9 @@ import java.util.Collection;
 
 @Entity
 @Table(name = "topology_host_request")
+@NamedQueries({
+  @NamedQuery(name = "TopologyHostRequestEntity.removeByIds", query = "DELETE FROM TopologyHostRequestEntity topologyHostRequest WHERE topologyHostRequest.id IN :hostRequestIds")
+})
 public class TopologyHostRequestEntity {
   @Id
 //  @GeneratedValue(strategy = GenerationType.TABLE, generator = "topology_host_request_id_generator")

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostTaskEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostTaskEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostTaskEntity.java
index 49d3a97..37830b7 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostTaskEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyHostTaskEntity.java
@@ -40,7 +40,11 @@ import java.util.Collection;
   pkColumnValue = "topology_host_task_id_seq", initialValue = 0)
 @NamedQueries({
   @NamedQuery(name = "TopologyHostTaskEntity.findByHostRequest",
-      query = "SELECT req FROM TopologyHostTaskEntity req WHERE req.topologyHostRequestEntity.id = :hostRequestId")
+      query = "SELECT req FROM TopologyHostTaskEntity req WHERE req.topologyHostRequestEntity.id = :hostRequestId"),
+  @NamedQuery(name = "TopologyLogicalTaskEntity.findHostRequestIdsByHostTaskIds",
+      query = "SELECT tht.hostRequestId from TopologyHostTaskEntity tht WHERE tht.id IN :hostTaskIds"),
+  @NamedQuery(name = "TopologyHostTaskEntity.removeByTaskIds",
+      query = "DELETE FROM TopologyHostTaskEntity tht WHERE tht.id IN :hostTaskIds")
 })
 public class TopologyHostTaskEntity {
   @Id
@@ -51,6 +55,9 @@ public class TopologyHostTaskEntity {
   @Column(name = "type", length = 255, nullable = false)
   private String type;
 
+  @Column(name = "host_request_id", nullable = false, insertable = false, updatable = false)
+  private Long hostRequestId;
+
   @ManyToOne
   @JoinColumn(name = "host_request_id", referencedColumnName = "id", nullable = false)
   private TopologyHostRequestEntity topologyHostRequestEntity;
@@ -67,7 +74,11 @@ public class TopologyHostTaskEntity {
   }
 
   public Long getHostRequestId() {
-    return topologyHostRequestEntity != null ? topologyHostRequestEntity.getId() : null;
+    return hostRequestId;
+  }
+
+  public void setHostRequestId(Long hostRequestId) {
+    this.hostRequestId = hostRequestId;
   }
 
   public String getType() {

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalRequestEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalRequestEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalRequestEntity.java
index 4d255b2..1536b80 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalRequestEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalRequestEntity.java
@@ -25,6 +25,8 @@ import javax.persistence.GenerationType;
 import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.OneToMany;
 import javax.persistence.OneToOne;
 import javax.persistence.Table;
@@ -33,6 +35,9 @@ import java.util.Collection;
 
 @Entity
 @Table(name = "topology_logical_request")
+@NamedQueries({
+  @NamedQuery(name = "TopologyLogicalRequestEntity.findRequestIds", query = "SELECT logicalrequest.topologyRequestId from TopologyLogicalRequestEntity logicalrequest WHERE logicalrequest.id IN :ids")
+})
 public class TopologyLogicalRequestEntity {
   @Id
   @Column(name = "id", nullable = false, updatable = false)

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalTaskEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalTaskEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalTaskEntity.java
index c71d4e4..2954863 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalTaskEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/TopologyLogicalTaskEntity.java
@@ -24,6 +24,8 @@ import javax.persistence.GenerationType;
 import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.OneToOne;
 import javax.persistence.Table;
 import javax.persistence.TableGenerator;
@@ -33,6 +35,10 @@ import javax.persistence.TableGenerator;
 @TableGenerator(name = "topology_logical_task_id_generator", table = "ambari_sequences",
   pkColumnName = "sequence_name", valueColumnName = "sequence_value",
   pkColumnValue = "topology_logical_task_id_seq", initialValue = 0)
+@NamedQueries({
+  @NamedQuery(name = "TopologyLogicalTaskEntity.findHostTaskIdsByPhysicalTaskIds", query = "SELECT logicaltask.hostTaskId from TopologyLogicalTaskEntity logicaltask WHERE logicaltask.physicalTaskId IN :physicalTaskIds"),
+  @NamedQuery(name = "TopologyLogicalTaskEntity.removeByPhysicalTaskIds", query = "DELETE FROM TopologyLogicalTaskEntity logicaltask WHERE logicaltask.physicalTaskId IN :taskIds")
+})
 public class TopologyLogicalTaskEntity {
   @Id
   @GeneratedValue(strategy = GenerationType.TABLE, generator = "topology_logical_task_id_generator")
@@ -42,12 +48,18 @@ public class TopologyLogicalTaskEntity {
   @Column(name = "component", length = 255)
   private String componentName;
 
+  @Column(name = "host_task_id", nullable = false, insertable = false, updatable = false)
+  private Long hostTaskId;
+
+  @Column(name = "physical_task_id", nullable = false, insertable = false, updatable = false)
+  private Long physicalTaskId;
+
   @ManyToOne
   @JoinColumn(name = "host_task_id", referencedColumnName = "id", nullable = false)
   private TopologyHostTaskEntity topologyHostTaskEntity;
 
   @OneToOne
-  @JoinColumn(name = "physical_task_id", referencedColumnName = "task_id")
+  @JoinColumn(name = "physical_task_id", referencedColumnName = "task_id", nullable = false)
   private HostRoleCommandEntity hostRoleCommandEntity;
 
   public Long getId() {
@@ -58,14 +70,22 @@ public class TopologyLogicalTaskEntity {
     this.id = id;
   }
 
-  public Long getHostTaskId() {
-    return topologyHostTaskEntity != null ? topologyHostTaskEntity.getId() : null;
-  }
-
   public Long getPhysicalTaskId() {
     return hostRoleCommandEntity != null ? hostRoleCommandEntity.getTaskId() : null;
   }
 
+  public void setPhysicalTaskId(Long physicalTaskId) {
+    this.physicalTaskId = physicalTaskId;
+  }
+
+  public void setHostTaskId(Long hostTaskId) {
+    this.hostTaskId = hostTaskId;
+  }
+
+  public Long getHostTaskId() {
+    return hostTaskId;
+  }
+
   public String getComponentName() {
     return componentName;
   }

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeEntity.java
index 0b27e3b..152fde1 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeEntity.java
@@ -65,6 +65,8 @@ import org.apache.ambari.server.state.stack.upgrade.UpgradeType;
       query = "SELECT u FROM UpgradeEntity u JOIN RequestEntity r ON u.requestId = r.requestId WHERE u.clusterId = :clusterId AND u.direction = :direction ORDER BY r.startTime DESC, u.upgradeId DESC"),
   @NamedQuery(name = "UpgradeEntity.findLatestForCluster",
       query = "SELECT u FROM UpgradeEntity u JOIN RequestEntity r ON u.requestId = r.requestId WHERE u.clusterId = :clusterId ORDER BY r.startTime DESC"),
+  @NamedQuery(name = "UpgradeEntity.findAllRequestIds",
+      query = "SELECT upgrade.requestId FROM UpgradeEntity upgrade")
 })
 public class UpgradeEntity {
 

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeItemEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeItemEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeItemEntity.java
index 560970a..35ea769 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeItemEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/UpgradeItemEntity.java
@@ -27,6 +27,8 @@ import javax.persistence.GenerationType;
 import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.ManyToOne;
+import javax.persistence.NamedQueries;
+import javax.persistence.NamedQuery;
 import javax.persistence.Table;
 import javax.persistence.TableGenerator;
 
@@ -48,6 +50,9 @@ import org.apache.ambari.server.state.UpgradeState;
     pkColumnValue = "upgrade_item_id_seq",
     initialValue = 0,
     allocationSize = 1000)
+@NamedQueries({
+  @NamedQuery(name = "UpgradeItemEntity.findAllStageIds", query = "SELECT upgradeItem.stageId FROM UpgradeItemEntity upgradeItem")
+})
 public class UpgradeItemEntity {
 
   @Id

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/python/ambari-server.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/python/ambari-server.py b/ambari-server/src/main/python/ambari-server.py
index 47256ab..235ef95 100755
--- a/ambari-server/src/main/python/ambari-server.py
+++ b/ambari-server/src/main/python/ambari-server.py
@@ -199,6 +199,12 @@ def restart(args):
   start(args)
 
 
+@OsFamilyFuncImpl(OsFamilyImpl.DEFAULT)
+def database_cleanup(args):
+  logger.info("Database cleanup.")
+  if args.silent:
+    stop(args)
+  db_cleanup(args)
 
 #
 # The Ambari Server status.
@@ -466,7 +472,7 @@ def init_parser_options(parser):
                     help="Print verbose status messages")
   parser.add_option("-s", "--silent",
                     action="store_true", dest="silent", default=False,
-                    help="Silently accepts default prompt values")
+                    help="Silently accepts default prompt values. For db-cleanup command, silent mode will stop ambari server.")
   parser.add_option('-g', '--debug', action="store_true", dest='debug', default=False,
                     help="Start ambari-server in debug mode")
   parser.add_option('-y', '--suspend-start', action="store_true", dest='suspend_start', default=False,
@@ -756,7 +762,7 @@ def create_user_action_map(args, options):
         CHECK_DATABASE_ACTION: UserAction(check_database, options),
         ENABLE_STACK_ACTION: UserAction(enable_stack, options, args),
         SETUP_SSO_ACTION: UserActionRestart(setup_sso, options),
-        DB_CLEANUP_ACTION: UserAction(db_cleanup, options),
+        DB_CLEANUP_ACTION: UserAction(database_cleanup, options),
         INSTALL_MPACK_ACTION: UserAction(install_mpack, options),
         UNINSTALL_MPACK_ACTION: UserAction(uninstall_mpack, options),
         UPGRADE_MPACK_ACTION: UserAction(upgrade_mpack, options),

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/main/python/ambari_server/dbCleanup.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/python/ambari_server/dbCleanup.py b/ambari-server/src/main/python/ambari_server/dbCleanup.py
index abc8267..6e16bc5 100644
--- a/ambari-server/src/main/python/ambari_server/dbCleanup.py
+++ b/ambari-server/src/main/python/ambari_server/dbCleanup.py
@@ -42,25 +42,29 @@ def run_db_cleanup(options):
     if validate_args(options):
         return 1
 
-    db_title = get_db_type(get_ambari_properties()).title
+    status, stateDesc = is_server_runing()
 
-    confirmBackup = get_YN_input("Ambari Server configured for {0}. Confirm you have made a backup of the Ambari Server database [y/n]".format(
-            db_title), True)
-    if not confirmBackup:
-        print_info_msg("Ambari Server Database cleanup aborted")
-        return 0
+    if not options.silent:
+      db_title = get_db_type(get_ambari_properties()).title
+
+      confirmBackup = get_YN_input("Ambari Server configured for {0}. Confirm you have made a backup of the Ambari Server database [y/n]".format(
+              db_title), True)
+      if not confirmBackup:
+          print_info_msg("Ambari Server Database cleanup aborted")
+          return 0
+
+      if status:
+          print_error_msg("The database cleanup cannot proceed while Ambari Server is running. Please shut down Ambari first.")
+          return 1
+
+      confirm = get_YN_input(
+          "Ambari server is using db type {0}. Cleanable database entries older than {1} will be cleaned up. Proceed [y/n]".format(
+              db_title, options.cleanup_from_date), True)
+      if not confirm:
+          print_info_msg("Ambari Server Database cleanup aborted")
+          return 0
 
-    status, stateDesc = is_server_runing()
-    if status:
-        print_error_msg("The database cleanup cannot proceed while Ambari Server is running. Please shut down Ambari first.")
-        return 1
 
-    confirm = get_YN_input(
-        "Ambari server is using db type {0}. Cleanable database entries older than {1} will be cleaned up. Proceed [y/n]".format(
-            db_title, options.cleanup_from_date), True)
-    if not confirm:
-        print_info_msg("Ambari Server Database cleanup aborted")
-        return 0
 
     jdk_path = get_java_exe_path()
     if jdk_path is None:
@@ -101,7 +105,6 @@ def run_db_cleanup(options):
 # Database cleanup
 #
 def db_cleanup(options):
-    logger.info("Database cleanup.")
     return run_db_cleanup(options)
 
 

http://git-wip-us.apache.org/repos/asf/ambari/blob/c23ef509/ambari-server/src/test/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelperTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelperTest.java b/ambari-server/src/test/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelperTest.java
index 5869630..9c8eb74 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelperTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/checks/DatabaseConsistencyCheckHelperTest.java
@@ -580,4 +580,70 @@ public class DatabaseConsistencyCheckHelperTest {
   }
 
 
+  @Test
+  public void testCheckForLargeTables() throws Exception {
+    EasyMockSupport easyMockSupport = new EasyMockSupport();
+    final AmbariMetaInfo mockAmbariMetainfo = easyMockSupport.createNiceMock(AmbariMetaInfo.class);
+    final DBAccessor mockDBDbAccessor = easyMockSupport.createNiceMock(DBAccessor.class);
+    final Connection mockConnection = easyMockSupport.createNiceMock(Connection.class);
+    final Statement mockStatement = easyMockSupport.createNiceMock(Statement.class);
+    final EntityManager mockEntityManager = easyMockSupport.createNiceMock(EntityManager.class);
+    final Clusters mockClusters = easyMockSupport.createNiceMock(Clusters.class);
+    final OsFamily mockOSFamily = easyMockSupport.createNiceMock(OsFamily.class);
+    final StackManagerFactory mockStackManagerFactory = easyMockSupport.createNiceMock(StackManagerFactory.class);
+
+    final ResultSet hostRoleCommandResultSet = easyMockSupport.createNiceMock(ResultSet.class);
+    final ResultSet executionCommandResultSet = easyMockSupport.createNiceMock(ResultSet.class);
+    final ResultSet stageResultSet = easyMockSupport.createNiceMock(ResultSet.class);
+    final ResultSet requestResultSet = easyMockSupport.createNiceMock(ResultSet.class);
+    final ResultSet alertHistoryResultSet = easyMockSupport.createNiceMock(ResultSet.class);
+
+    final Injector mockInjector = Guice.createInjector(new AbstractModule() {
+      @Override
+      protected void configure() {
+        bind(AmbariMetaInfo.class).toInstance(mockAmbariMetainfo);
+        bind(StackManagerFactory.class).toInstance(mockStackManagerFactory);
+        bind(EntityManager.class).toInstance(mockEntityManager);
+        bind(DBAccessor.class).toInstance(mockDBDbAccessor);
+        bind(Clusters.class).toInstance(mockClusters);
+        bind(OsFamily.class).toInstance(mockOSFamily);
+      }
+    });
+
+    expect(hostRoleCommandResultSet.next()).andReturn(true).once();
+    expect(executionCommandResultSet.next()).andReturn(true).once();
+    expect(stageResultSet.next()).andReturn(true).once();
+    expect(requestResultSet.next()).andReturn(true).once();
+    expect(alertHistoryResultSet.next()).andReturn(true).once();
+    expect(hostRoleCommandResultSet.getLong(1)).andReturn(2345L).atLeastOnce();
+    expect(executionCommandResultSet.getLong(1)).andReturn(12345L).atLeastOnce();
+    expect(stageResultSet.getLong(1)).andReturn(2321L).atLeastOnce();
+    expect(requestResultSet.getLong(1)).andReturn(1111L).atLeastOnce();
+    expect(alertHistoryResultSet.getLong(1)).andReturn(2223L).atLeastOnce();
+    expect(mockDBDbAccessor.getConnection()).andReturn(mockConnection);
+    expect(mockDBDbAccessor.getDbType()).andReturn(DBAccessor.DbType.MYSQL);
+    expect(mockDBDbAccessor.getDbSchema()).andReturn("test_schema");
+    expect(mockConnection.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE)).andReturn(mockStatement).anyTimes();
+    expect(mockStatement.executeQuery("SELECT (data_length + index_length) \"Table Size\" " +
+            "FROM information_schema.TABLES WHERE table_schema = \"test_schema\" AND table_name =\"host_role_command\"")).andReturn(hostRoleCommandResultSet);
+    expect(mockStatement.executeQuery("SELECT (data_length + index_length) \"Table Size\" " +
+            "FROM information_schema.TABLES WHERE table_schema = \"test_schema\" AND table_name =\"execution_command\"")).andReturn(executionCommandResultSet);
+    expect(mockStatement.executeQuery("SELECT (data_length + index_length) \"Table Size\" " +
+            "FROM information_schema.TABLES WHERE table_schema = \"test_schema\" AND table_name =\"stage\"")).andReturn(stageResultSet);
+    expect(mockStatement.executeQuery("SELECT (data_length + index_length) \"Table Size\" " +
+            "FROM information_schema.TABLES WHERE table_schema = \"test_schema\" AND table_name =\"request\"")).andReturn(requestResultSet);
+    expect(mockStatement.executeQuery("SELECT (data_length + index_length) \"Table Size\" " +
+            "FROM information_schema.TABLES WHERE table_schema = \"test_schema\" AND table_name =\"alert_history\"")).andReturn(alertHistoryResultSet);
+
+    DatabaseConsistencyCheckHelper.setInjector(mockInjector);
+
+    easyMockSupport.replayAll();
+
+    mockAmbariMetainfo.init();
+
+    DatabaseConsistencyCheckHelper.resetCheckResult();
+    DatabaseConsistencyCheckHelper.checkForLargeTables();
+
+    easyMockSupport.verifyAll();
+  }
 }


Mime
View raw message