// for each s, we may need to create a new "chunk"
// each partition is created by DFS from a source
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;
    }
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
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);
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();