[HOPSWORKS-CLOUD-1540] Fixing cleaning up subtree operations

Author: smkniazi

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

@@ -25,6 +25,7 @@
 import io.hops.exception.StorageException;
 import io.hops.metadata.hdfs.TablesDef;
 import io.hops.metadata.hdfs.dal.OngoingSubTreeOpsDataAccess;
+import io.hops.metadata.hdfs.entity.Storage;
 import io.hops.metadata.hdfs.entity.SubTreeOperation;
 import io.hops.metadata.ndb.ClusterjConnector;
 import io.hops.metadata.ndb.wrapper.HopsPredicate;
@@ -35,6 +36,7 @@
 import io.hops.metadata.ndb.wrapper.HopsSession;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
@@ -149,6 +151,90 @@ public Collection<SubTreeOperation> allOpsToRecoverAsync()
     return convertAndRelease(dbSession, query.getResultList());
   }
 
+  @Override
+  public Collection<SubTreeOperation> allDeadOperations(long[] aliveNNIDs, long time) throws StorageException {
+    if(aliveNNIDs == null || aliveNNIDs.length == 0){
+      throw new IllegalArgumentException("No alive namenode specified");
+    }
+
+    HopsSession dbSession = connector.obtainSession();
+    HopsQueryBuilder qb = dbSession.getQueryBuilder();
+    HopsQueryDomainType dobj = qb.createQueryDefinition(OnGoingSubTreeOpsDTO.class);
+
+    HopsPredicate pred[] = new HopsPredicate[aliveNNIDs.length];
+    for(int i = 0; i < aliveNNIDs.length; i++ ){
+      pred[i] = dobj.not(dobj.get("namenodeId").equal(dobj.param("namenodeIdParam"+i)));
+    }
+
+    HopsPredicate allCombined = pred[0];
+    for(int i = 1; i < aliveNNIDs.length; i++ ){
+      allCombined = allCombined.and(pred[i]);
+    }
+    allCombined.and(dobj.get("startTime").lessThan(dobj.param("startTimeParam")));
+
+    dobj.where(allCombined);
+    HopsQuery query = dbSession.createQuery(dobj);
+
+    for(int i = 0; i < aliveNNIDs.length; i++ ){
+      query.setParameter("namenodeIdParam"+i, aliveNNIDs[i]);
+    }
+    query.setParameter("startTimeParam", time);
+
+    return convertAndRelease(dbSession, query.getResultList());
+  }
+
+  @Override
+  public Collection<SubTreeOperation> allSlowActiveOperations(long[] aliveNNIDs, long time)
+          throws StorageException {
+    if(aliveNNIDs == null || aliveNNIDs.length == 0){
+      throw new IllegalArgumentException("No alive namenode specified");
+    }
+
+    HopsSession dbSession = connector.obtainSession();
+    HopsQueryBuilder qb = dbSession.getQueryBuilder();
+    HopsQueryDomainType dobj = qb.createQueryDefinition(OnGoingSubTreeOpsDTO.class);
+
+    HopsPredicate pred[] = new HopsPredicate[aliveNNIDs.length];
+    for(int i = 0; i < aliveNNIDs.length; i++ ){
+      pred[i] = dobj.get("namenodeId").equal(dobj.param("namenodeIdParam"+i));
+    }
+
+    HopsPredicate allCombined = pred[0];
+    for(int i = 1; i < aliveNNIDs.length; i++ ){
+      allCombined = allCombined.or(pred[i]);
+    }
+    allCombined = dobj.get("startTime").lessThan(dobj.param("startTimeParam")).and(allCombined);
+
+    dobj.where(allCombined);
+    HopsQuery query = dbSession.createQuery(dobj);
+
+    for(int i = 0; i < aliveNNIDs.length; i++ ){
+      query.setParameter("namenodeIdParam"+i, aliveNNIDs[i]);
+    }
+    query.setParameter("startTimeParam", time);
+
+    return convertAndRelease(dbSession, query.getResultList());
+  }
+
+  @Override
+  public long getLockTime(long inodeID) throws StorageException {
+    HopsSession dbSession = connector.obtainSession();
+    HopsQueryBuilder qb = dbSession.getQueryBuilder();
+    HopsQueryDomainType dobj = qb.createQueryDefinition(OnGoingSubTreeOpsDTO.class);
+    HopsPredicate pred = dobj.get("inodeId").equal(dobj.param("inodeIdParam"));
+    dobj.where(pred);
+    HopsQuery query = dbSession.createQuery(dobj);
+    query.setParameter("inodeIdParam", inodeID);
+    List<SubTreeOperation> list = convertAndRelease(dbSession, query.getResultList());
+    if (list.size() > 0) {
+      if (list.size() > 1) {
+        throw new StorageException("Multiple subtree locks found for same INode: " + inodeID);
+      }
+      return list.get(0).getStartTime();
+    }
+    return 0;
+  }
+
   @Override
   public Collection<SubTreeOperation> allOpsByNN(long nnID)
           throws StorageException {