[HOPS-1655] fixed quota updates for delete, rename and overwrite operations

Author: smkniazi

src/main/java/io/hops/metadata/ndb/dalimpl/hdfs/InvalidatedBlockClusterj.java

@@ -28,6 +28,7 @@
 import io.hops.exception.StorageException;
 import io.hops.metadata.hdfs.TablesDef;
 import io.hops.metadata.hdfs.dal.InvalidateBlockDataAccess;
+import io.hops.metadata.hdfs.dal.SQLResultSetHandler;
 import io.hops.metadata.hdfs.entity.InvalidatedBlock;
 import io.hops.metadata.ndb.ClusterjConnector;
 import io.hops.metadata.ndb.mysqlserver.MySQLQueryHelper;
@@ -121,7 +122,8 @@ public List<InvalidatedBlock> findInvalidatedBlockByStorageId(int storageId)
   @Override
   public Map<Long, Long> findInvalidatedBlockBySidUsingMySQLServer(int storageId) throws StorageException {
     return MySQLQueryHelper.execute(String.format("SELECT %s, %s "
-            + "FROM %s WHERE %s='%d'", BLOCK_ID, GENERATION_STAMP, TABLE_NAME, STORAGE_ID, storageId), new MySQLQueryHelper.ResultSetHandler<Map<Long,Long>>() {
+            + "FROM %s WHERE %s='%d'", BLOCK_ID, GENERATION_STAMP, TABLE_NAME, STORAGE_ID,
+      storageId), new SQLResultSetHandler<Map<Long,Long>>() {
       @Override
       public Map<Long,Long> handle(ResultSet result) throws SQLException {
         Map<Long,Long> blockInodeMap = new HashMap<>();

src/main/java/io/hops/metadata/ndb/dalimpl/hdfs/QuotaUpdateClusterj.java

@@ -24,6 +24,7 @@
 import io.hops.exception.StorageException;
 import io.hops.metadata.hdfs.TablesDef;
 import io.hops.metadata.hdfs.dal.QuotaUpdateDataAccess;
+import io.hops.metadata.hdfs.dal.SQLResultSetHandler;
 import io.hops.metadata.hdfs.entity.QuotaUpdate;
 import io.hops.metadata.ndb.ClusterjConnector;
 import io.hops.metadata.ndb.mysqlserver.HopsSQLExceptionHelper;
@@ -41,6 +42,7 @@
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -244,7 +246,7 @@ private QuotaUpdate convertAndRelease(HopsSession session, QuotaUpdateDTO dto) t
   private static final String INODE_ID_PARAM = "inodeId";
 
   @Override
-  public List<QuotaUpdate> findByInodeId(long inodeId) throws StorageException {
+  public List<QuotaUpdate> findByInodeId(final long inodeId, final int limit) throws StorageException {
     HopsSession session = connector.obtainSession();
     HopsQueryBuilder qb = session.getQueryBuilder();
     HopsQueryDomainType<QuotaUpdateDTO> dobj =
@@ -254,6 +256,10 @@ public List<QuotaUpdate> findByInodeId(long inodeId) throws StorageException {
     HopsQuery<QuotaUpdateDTO> query = session.createQuery(dobj);
     query.setParameter(INODE_ID_PARAM, inodeId);
 
+    if (limit != -1) {
+      query.setLimits(0, limit);
+    }
+
     List<QuotaUpdateDTO> results = query.getResultList();
     return convertAndRelease(session, results);
   }
@@ -263,4 +269,20 @@ public int getCount() throws StorageException {
     int count = MySQLQueryHelper.countAll(TablesDef.QuotaUpdateTableDef.TABLE_NAME);
     return count;
   }
+
+  @Override
+  public List<Long> getDistinctInodes() throws StorageException {
+    final String query = "select DISTINCT(" + INODE_ID + ") from " + TABLE_NAME;
+    return MySQLQueryHelper.execute(query, new SQLResultSetHandler<List<Long>>() {
+      @Override
+      public List<Long> handle(ResultSet result) throws SQLException {
+        List<Long> inodes = new ArrayList<>();
+
+        while (result.next()){
+          inodes.add(result.getLong(INODE_ID));
+        }
+        return inodes;
+      }
+    });
+  }
 }

src/main/java/io/hops/metadata/ndb/dalimpl/hdfs/ReplicaClusterj.java

@@ -29,6 +29,7 @@
 import io.hops.exception.StorageException;
 import io.hops.metadata.hdfs.TablesDef;
 import io.hops.metadata.hdfs.dal.ReplicaDataAccess;
+import io.hops.metadata.hdfs.dal.SQLResultSetHandler;
 import io.hops.metadata.hdfs.entity.Replica;
 import io.hops.metadata.ndb.ClusterjConnector;
 import io.hops.metadata.ndb.mysqlserver.MySQLQueryHelper;
@@ -239,7 +240,7 @@ protected static Set<Long> getReplicas(int storageId) throws
       StorageException {
     return MySQLQueryHelper.execute(String.format("SELECT %s " +
         "FROM %s WHERE %s='%d'", BLOCK_ID, TABLE_NAME, STORAGE_ID, storageId)
-        , new MySQLQueryHelper.ResultSetHandler<Set<Long>>() {
+        , new SQLResultSetHandler<Set<Long>>() {
       @Override
       public Set<Long> handle(ResultSet result) throws SQLException {
         Set<Long> blocks = Sets.newHashSet();

src/main/java/io/hops/metadata/ndb/mysqlserver/MySQLQueryHelper.java

@@ -24,6 +24,8 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+
+import io.hops.metadata.hdfs.dal.SQLResultSetHandler;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -125,7 +127,7 @@ public static long maxLong(String tableName, String column)
 
   public static int executeIntAggrQuery(final String query)
       throws StorageException {
-    return execute(query, new ResultSetHandler<Integer>() {
+    return execute(query, new SQLResultSetHandler<Integer>() {
       @Override
       public Integer handle(ResultSet result) throws SQLException, StorageException {
         if (!result.next()) {
@@ -139,7 +141,7 @@ public Integer handle(ResultSet result) throws SQLException, StorageException {
 
     public static long executeLongAggrQuery(final String query)
       throws StorageException {
-    return execute(query, new ResultSetHandler<Long>() {
+    return execute(query, new SQLResultSetHandler<Long>() {
       @Override
       public Long handle(ResultSet result) throws SQLException, StorageException {
         if (!result.next()) {
@@ -153,7 +155,7 @@ public Long handle(ResultSet result) throws SQLException, StorageException {
 
   private static boolean executeBooleanQuery(final String query)
       throws StorageException {
-    return execute(query, new ResultSetHandler<Boolean>() {
+    return execute(query, new SQLResultSetHandler<Boolean>() {
       @Override
       public Boolean handle(ResultSet result) throws SQLException, StorageException {
         if (!result.next()) {
@@ -185,11 +187,7 @@ public static int execute(String query) throws StorageException {
     }
   }
 
-  public interface ResultSetHandler<R> {
-    R handle(ResultSet result) throws SQLException, StorageException;
-  }
-
-  public static <R> R execute(String query, ResultSetHandler<R> handler)
+  public static <R> R execute(String query, SQLResultSetHandler<R> handler)
       throws StorageException {
     try {
       PreparedStatement s = null;

src/main/java/io/hops/metadata/ndb/mysqlserver/MysqlServerConnector.java

@@ -29,6 +29,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+
+import io.hops.metadata.hdfs.dal.SQLResultSetHandler;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import java.sql.Connection;
@@ -301,7 +303,7 @@ public String getDatabaseName() {
   public static float getResourceMemUtilization() throws StorageException {
     return MySQLQueryHelper.execute("SELECT memory_type, used, total FROM " +
                     "ndbinfo.memoryusage where memory_type = \"Data memory\"",
-            new MySQLQueryHelper.ResultSetHandler<Float>() {
+            new SQLResultSetHandler<Float>() {
               @Override
               public Float handle(ResultSet result)
                       throws SQLException {
@@ -318,7 +320,7 @@ public Float handle(ResultSet result)
   public static boolean hasResources(final double threshold) throws StorageException {
     return MySQLQueryHelper.execute("SELECT memory_type, used, total FROM " +
             "ndbinfo.memoryusage where memory_type = \"Data memory\"",
-        new MySQLQueryHelper.ResultSetHandler<Boolean>() {
+        new SQLResultSetHandler<Boolean>() {
           @Override
           public Boolean handle(ResultSet result)
               throws SQLException, StorageException {