public List<PigOutLogicalChunks> createLogicalChunks() throws FrontendException {
        Set<Operator> seenByDFS = new HashSet<Operator>();
        Map<LogicalPlan, PigOutLogicalChunks> mapping =
                new HashMap<>();
        while ( !visitQueue.isEmpty() ) {
            Operator s = visitQueue.poll();
            LogicalRelationalOperator lro = (LogicalRelationalOperator) s;
            log.debug("Visit: " + lro);
            // for each s, we may need to create a new "chunk"
			// each partition is created by DFS from a source
            if (!seenByDFS.add(s)) continue;
            // Create a new logical plan for the new chunk
            LogicalPlan newPlan = new LogicalPlan();
            newPlan.add(s);
            subPlans.put(s, newPlan);
            List<Operator> preds = this.plan.getPredecessors(s);
            if (preds != null)
                planPreds.put(newPlan, new LinkedList<>(preds));
            log.debug(lro.getAlias() + " is the root of new chunk");
            discoverPartition(s, seenByDFS, newPlan);
            String nameNode = this.siteAssignment.get(s);
            String hostname = nameNode.split(":")[0];
            String portNum = this.context.getProperties().getProperty(hostname + ".job.tracker").split(":")[1];
            int port = new Integer(portNum);
            PigOutCluster clusterInfo = getPigOutCluster(hostname, port);
            long dataIn = computeDataInSize(newPlan);
            long dataOut = computeDataOutSize(newPlan);
            addStores(newPlan);
            addLoads(newPlan);
            Iterator<Operator> it = newPlan.getOperators();
            while (it.hasNext()) {
                Operator op = it.next();
                op.setPlan(newPlan);
            }
            PigOutLogicalChunks chunk;
            chunk = new PigOutLogicalChunks(newPlan, context, clusterInfo, dataIn, dataOut);
            chunk.setAlias(lro.getAlias());
            mapping.put( newPlan, chunk );
        }
        for (LogicalPlan lp : mapping.keySet()) {
            List<Operator> succs = planSuccs.get(lp);
            List<Operator> preds = planPreds.get(lp);
            PigOutLogicalChunks chunk = mapping.get(lp);
            if (succs != null) {
                for (Operator succ : succs) {
                    LogicalPlan succLP = (LogicalPlan) succ.getPlan();
                    PigOutLogicalChunks succChunk = mapping.get(succLP);
                    succChunk.addPredecessors(chunk);
                    log.debug(succChunk.getAlias() + " depends on " + chunk.getAlias());
                }
            }
            if (preds != null) {
                for (Operator pred : preds) {
                    LogicalPlan predLP = (LogicalPlan) pred.getPlan();
                    PigOutLogicalChunks predChunk = mapping.get(predLP);
                    chunk.addPredecessors(predChunk);
                    log.debug(chunk.getAlias() + " depends on " + predChunk.getAlias());
                }
            }
        }
        List<PigOutLogicalChunks> list_chunks = new LinkedList<>();
        for (PigOutLogicalChunks chunk : mapping.values()) {
            list_chunks.add( chunk );
        }
        return list_chunks;
    }
public List<PigOutLogicalChunks> createLogicalChunks() throws FrontendException {
        Set<Operator> seenByDFS = new HashSet<Operator>();
        Map<LogicalPlan, PigOutLogicalChunks> mapping =
                new HashMap<>();
        while ( !visitQueue.isEmpty() ) {
            Operator s = visitQueue.poll();
            LogicalRelationalOperator lro = (LogicalRelationalOperator) s;
            log.debug("Visit: " + lro);
           
		
            if (!seenByDFS.add(s)) continue;
           
            LogicalPlan newPlan = new LogicalPlan();
            newPlan.add(s);
            subPlans.put(s, newPlan);
            List<Operator> preds = this.plan.getPredecessors(s);
            if (preds != null)
                planPreds.put(newPlan, new LinkedList<>(preds));
            log.debug(lro.getAlias() + " is the root of new chunk");
            discoverPartition(s, seenByDFS, newPlan);
            String nameNode = this.siteAssignment.get(s);
            String hostname = nameNode.split(":")[0];
            String portNum = this.context.getProperties().getProperty(hostname + ".job.tracker").split(":")[1];
            int port = new Integer(portNum);
            PigOutCluster clusterInfo = getPigOutCluster(hostname, port);
            long dataIn = computeDataInSize(newPlan);
            long dataOut = computeDataOutSize(newPlan);
            addStores(newPlan);
            addLoads(newPlan);
            Iterator<Operator> it = newPlan.getOperators();
            while (it.hasNext()) {
                Operator op = it.next();
                op.setPlan(newPlan);
            }
            PigOutLogicalChunks chunk;
            chunk = new PigOutLogicalChunks(newPlan, context, clusterInfo, dataIn, dataOut);
            chunk.setAlias(lro.getAlias());
            mapping.put( newPlan, chunk );
        }
        for (LogicalPlan lp : mapping.keySet()) {
            List<Operator> succs = planSuccs.get(lp);
            List<Operator> preds = planPreds.get(lp);
            PigOutLogicalChunks chunk = mapping.get(lp);
            if (succs != null) {
                for (Operator succ : succs) {
                    LogicalPlan succLP = (LogicalPlan) succ.getPlan();
                    PigOutLogicalChunks succChunk = mapping.get(succLP);
                    succChunk.addPredecessors(chunk);
                    log.debug(succChunk.getAlias() + " depends on " + chunk.getAlias());
                }
            }
            if (preds != null) {
                for (Operator pred : preds) {
                    LogicalPlan predLP = (LogicalPlan) pred.getPlan();
                    PigOutLogicalChunks predChunk = mapping.get(predLP);
                    chunk.addPredecessors(predChunk);
                    log.debug(chunk.getAlias() + " depends on " + predChunk.getAlias());
                }
            }
        }
        List<PigOutLogicalChunks> list_chunks = new LinkedList<>();
        for (PigOutLogicalChunks chunk : mapping.values()) {
            list_chunks.add( chunk );
        }
        return list_chunks;
    }
public list<pigoutlogicalchunks> createlogicalchunks() throws frontendexception { set<operator> seenbydfs = new hashset<operator>(); map<logicalplan, pigoutlogicalchunks> mapping = new hashmap<>(); while ( !visitqueue.isempty() ) { operator s = visitqueue.poll(); logicalrelationaloperator lro = (logicalrelationaloperator) s; log.debug("visit: " + lro); if (!seenbydfs.add(s)) continue; logicalplan newplan = new logicalplan(); newplan.add(s); subplans.put(s, newplan); list<operator> preds = this.plan.getpredecessors(s); if (preds != null) planpreds.put(newplan, new linkedlist<>(preds)); log.debug(lro.getalias() + " is the root of new chunk"); discoverpartition(s, seenbydfs, newplan); string namenode = this.siteassignment.get(s); string hostname = namenode.split(":")[0]; string portnum = this.context.getproperties().getproperty(hostname + ".job.tracker").split(":")[1]; int port = new integer(portnum); pigoutcluster clusterinfo = getpigoutcluster(hostname, port); long datain = computedatainsize(newplan); long dataout = computedataoutsize(newplan); addstores(newplan); addloads(newplan); iterator<operator> it = newplan.getoperators(); while (it.hasnext()) { operator op = it.next(); op.setplan(newplan); } pigoutlogicalchunks chunk; chunk = new pigoutlogicalchunks(newplan, context, clusterinfo, datain, dataout); chunk.setalias(lro.getalias()); mapping.put( newplan, chunk ); } for (logicalplan lp : mapping.keyset()) { list<operator> succs = plansuccs.get(lp); list<operator> preds = planpreds.get(lp); pigoutlogicalchunks chunk = mapping.get(lp); if (succs != null) { for (operator succ : succs) { logicalplan succlp = (logicalplan) succ.getplan(); pigoutlogicalchunks succchunk = mapping.get(succlp); succchunk.addpredecessors(chunk); log.debug(succchunk.getalias() + " depends on " + chunk.getalias()); } } if (preds != null) { for (operator pred : preds) { logicalplan predlp = (logicalplan) pred.getplan(); pigoutlogicalchunks predchunk = mapping.get(predlp); chunk.addpredecessors(predchunk); log.debug(chunk.getalias() + " depends on " + predchunk.getalias()); } } } list<pigoutlogicalchunks> list_chunks = new linkedlist<>(); for (pigoutlogicalchunks chunk : mapping.values()) { list_chunks.add( chunk ); } return list_chunks; }