/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.WeakHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.ExpressionOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.InternalCachedBag;
import org.apache.pig.data.SelfSpillBag.MemoryLimits;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.GroupingSpillable;
import org.apache.pig.impl.util.Spillable;
import org.apache.pig.impl.util.SpillableMemoryManager;

import com.google.common.collect.Maps;

/**
 * Do partial aggregation in map plan. Inputs are buffered up in
 * a hashmap until a threshold is reached; then the combiner functions
 * are fed these buffered up inputs, and results stored in a secondary
 * map. Once that map fills up or all input has been seen, results are
 * piped out into the next operator (caller of getNext()).
 */
public class POPartialAgg extends PhysicalOperator implements Spillable, GroupingSpillable {
    private static final Log LOG = LogFactory.getLog(POPartialAgg.class);
    private static final long serialVersionUID = 1L;

    private static final Result EOP_RESULT = new Result(POStatus.STATUS_EOP,
            null);

    // number of records to sample to determine average size used by each
    // entry in hash map and average seen reduction
    private static final int NUM_RECS_TO_SAMPLE = 10000;

    // We want to allow bigger list sizes for group all.
    // But still have a cap on it to avoid JVM finding it hard to allocate space
    // TODO: How high can we go without performance degradation??
    private static final int MAX_LIST_SIZE = 25000;

    // We want to avoid massive ArrayList copies as they get big.
    // Array Lists grow by prevSize + prevSize/2. Given default initial size of 10,
    // 9369 is the size of the array after 18 such resizings. This seems like a sufficiently
    // large value to trigger spilling/aggregation instead of paying for yet another data
    // copy.
    // For group all cases, we will set this to a higher value
    private int listSizeThreshold = 9367;

    // Using default min reduction 7 instead of 10 as processedInputMap size
    // will be 4096 (hashmap size is power of 2) for both 20000/10 and 20000/7.
    // So using the lower number 7 as even 7x reduction is worth using map side aggregation
    private static final int DEFAULT_MIN_REDUCTION = 7;

    // TODO: these are temporary. The real thing should be using memory usage estimation.
    private static final int FIRST_TIER_THRESHOLD = 20000;
    private static final int SECOND_TIER_THRESHOLD = FIRST_TIER_THRESHOLD / DEFAULT_MIN_REDUCTION;

    private static final WeakHashMap<POPartialAgg, Byte> ALL_POPARTS = new WeakHashMap<POPartialAgg, Byte>();

    private PhysicalPlan keyPlan;
    private ExpressionOperator keyLeaf;
    private List<PhysicalPlan> valuePlans;
    private List<ExpressionOperator> valueLeaves;
    private boolean isGroupAll;

    private transient int numRecsInRawMap;
    private transient int numRecsInProcessedMap;

    private transient Map<Object, List<Tuple>> rawInputMap;
    private transient Map<Object, List<Tuple>> processedInputMap;

    //Transient booleans always initialize to false
    private transient boolean initialized;
    private transient boolean disableMapAgg;
    private transient boolean sizeReductionChecked;
    private transient boolean inputsExhausted;
    private transient boolean estimatedMemThresholds;
    // The doSpill flag is set when spilling is running or needs to run.
    // It is set by POPartialAgg when its buffers are full after having run aggregations and
    // the records have to be emitted to the map output.
    // The doContingentSpill flag is set when the SpillableMemoryManager is notified
    // by GC that the runtime is low on memory and the SpillableMemoryManager identifies
    // the particular buffer as a good spill candidate because it is large. The contingent spill logic tries
    // to satisfy the memory manager's request for freeing memory by aggregating data
    // rather than just spilling records to disk.
    private transient volatile boolean doSpill;
    private transient volatile boolean doContingentSpill;
    private transient volatile boolean startedContingentSpill;
    private transient volatile Object spillLock;

    private transient int minOutputReduction;
    private transient float percentUsage;
    private transient int numRecordsToSample;
    private transient int firstTierThreshold;
    private transient int secondTierThreshold;
    private transient int sizeReduction;
    private transient int avgTupleSize;
    private transient Iterator<Entry<Object, List<Tuple>>> spillingIterator;

    public POPartialAgg(OperatorKey k) {
        this(k, false);
    }

    public POPartialAgg(OperatorKey k, boolean isGroupAll) {
        super(k);
        this.isGroupAll = isGroupAll;
    }

    private void init() throws ExecException {

        numRecsInRawMap = 0;
        numRecsInProcessedMap = 0;
        minOutputReduction = DEFAULT_MIN_REDUCTION;
        numRecordsToSample = NUM_RECS_TO_SAMPLE;
        firstTierThreshold = FIRST_TIER_THRESHOLD;
        secondTierThreshold = SECOND_TIER_THRESHOLD;
        sizeReduction = 1;
        avgTupleSize = 0;
        percentUsage = 0.2F;
        spillLock = new Object();
        if (PigMapReduce.sJobConfInternal.get() != null) {
            String usage = PigMapReduce.sJobConfInternal.get().get(
                    PigConfiguration.PIG_CACHEDBAG_MEMUSAGE);
            if (usage != null) {
                percentUsage = Float.parseFloat(usage);
            }
            minOutputReduction = PigMapReduce.sJobConfInternal.get().getInt(
                    PigConfiguration.PIG_EXEC_MAP_PARTAGG_MINREDUCTION, DEFAULT_MIN_REDUCTION);
            if (minOutputReduction <= 0) {
                LOG.info("Specified reduction is < 0 (" + minOutputReduction + "). Using default " +
                        DEFAULT_MIN_REDUCTION);
                minOutputReduction = DEFAULT_MIN_REDUCTION;
            }
        }
        if (percentUsage <= 0) {
            LOG.info("No memory allocated to intermediate memory buffers. Turning off partial aggregation.");
            disableMapAgg = true;
            // Set them to true instead of adding another check for !disableMapAgg
            sizeReductionChecked = true;
            estimatedMemThresholds = true;
        } else {
            ALL_POPARTS.put(this, null);
            SpillableMemoryManager.getInstance().registerSpillable(this);
        }
        // Avoid hashmap resizing. TODO: Investigate loadfactor of 0.90 or 1.0
        // newHashMapWithExpectedSize does new HashMap(expectedSize + expectedSize/3)
        // to factor in default load factor of 0.75.
        // For Hashmap, internally its size is always in power of 2.
        // So for NUM_RECS_TO_SAMPLE=10000, hashmap size will be 16384
        // With secondTierThreshold of 2857 (minReduction 7), hashmap size will be 4096
        if (!disableMapAgg) {
            rawInputMap = Maps.newHashMapWithExpectedSize(NUM_RECS_TO_SAMPLE);
            processedInputMap = Maps.newHashMapWithExpectedSize(SECOND_TIER_THRESHOLD);
        }
        if (isGroupAll) {
            listSizeThreshold = Math.min(numRecordsToSample, MAX_LIST_SIZE);
        }
        initialized = true;
    }

    @Override
    public Result getNextTuple() throws ExecException {
        // accumulate tuples from processInput in rawInputMap.
        // when the maps grow to mem limit, go over each item in map, and call
        // combiner aggs on each collection.
        // Store the results into processedInputMap. Clear out rawInputMap.
        // Mem usage is updated every time we modify either of the maps.
        // When processedInputMap is >= 20% of allotted memory, run aggs on it,
        // and output the results as returns of successive calls of this method.
        // Then reset processedInputMap.
        // The fact that we are in the latter stage is communicated via the doSpill
        // flag.

        if (!initialized) {
            init();
        }

        while (true) {
            if (!sizeReductionChecked && numRecsInRawMap >= numRecordsToSample) {
                checkSizeReduction();
                if (doContingentSpill && !doSpill) {
                    LOG.info("Avoided emitting records during spill memory call.");
                    doContingentSpill = false;
                }
            }
            if (!estimatedMemThresholds && numRecsInRawMap >= numRecordsToSample) {
                estimateMemThresholds();
            }
            if (doContingentSpill) {
                startedContingentSpill = true;
                // Don't aggregate if spilling. Avoid concurrent update of spilling iterator.
                if (doSpill == false) {
                    // SpillableMemoryManager requested a spill to reduce memory
                    // consumption. See if we can avoid it.
                    // If we see that there are way more records in processed map than
                    // raw map, it is better to spill after the aggregation
                    boolean spillProcessedMap = (3 * numRecsInRawMap) < numRecsInProcessedMap;
                    aggregateBothLevels(false, false);
                    // Spill if the numRecsInProcessedMap
                    if (spillProcessedMap || shouldSpill()) {
                        startSpill(false);
                    } else {
                        LOG.info("Avoided emitting records during spill memory call.");
                        doContingentSpill = false;
                    }
                }
            }
            if (doSpill) {
                startSpill(true);
                Result result = spillResult();
                if (result.returnStatus == POStatus.STATUS_EOP) {
                    doSpill = false;
                    doContingentSpill = false;
                }
                if (result.returnStatus != POStatus.STATUS_EOP) {
                    return result;
                } else if (inputsExhausted) {
                    freeMemory();
                    return result;
                }
            }
            if (mapAggDisabled()) {
                // disableMapAgg() sets doSpill, so we can't get here while there is still contents in the buffered maps.
                // if we get to this point, everything is flushed, so we can simply return the raw tuples from now on.
                freeMemory();
                return processInput();
            } else {
                Result inp = processInput();
                if (inp.returnStatus == POStatus.STATUS_ERR) {
                    return inp;
                } else if (inp.returnStatus == POStatus.STATUS_EOP) {
                    if (parentPlan.endOfAllInput) {
                        // parent input is over. flush what we have.
                        inputsExhausted = true;
                        LOG.info("Spilling last bits.");
                        startSpill(true);
                        continue;
                    } else {
                        return EOP_RESULT;
                    }
                } else if (inp.returnStatus == POStatus.STATUS_NULL) {
                    continue;
                } else {
                    // add this input to map.
                    Tuple inpTuple = (Tuple) inp.result;
                    keyPlan.attachInput(inpTuple);

                    // evaluate the key
                    Result keyRes = getResult(keyLeaf);
                    if (keyRes.returnStatus != POStatus.STATUS_OK) {
                        return keyRes;
                    }
                    Object key = keyRes.result;
                    keyPlan.detachInput();
                    numRecsInRawMap += 1;
                    addKeyValToMap(rawInputMap, key, inpTuple);

                    aggregateBothLevels(true, true);
                    if (shouldSpill()) {
                        startSpill(false); // next time around, we'll start emitting.
                    }
                }
            }
        }
    }

    private void freeMemory() throws ExecException {
        if (rawInputMap != null && !rawInputMap.isEmpty()) {
            throw new ExecException("Illegal state. Trying to free up partial aggregation maps when they are not empty");
        }
        // Free up the maps for garbage collection
        rawInputMap = null;
        processedInputMap = null;
    }

    private void estimateMemThresholds() {
        if (!mapAggDisabled()) {
            LOG.info("Getting mem limits; considering " + ALL_POPARTS.size()
                    + " POPArtialAgg objects." + " with memory percentage "
                    + percentUsage);
            MemoryLimits memLimits = new MemoryLimits(ALL_POPARTS.size(), percentUsage);
            int estTotalMem = 0;
            int estTuples = 0;
            for (Map.Entry<Object, List<Tuple>> entry : rawInputMap.entrySet()) {
                for (Tuple t : entry.getValue()) {
                    estTuples += 1;
                    int mem = (int) t.getMemorySize();
                    estTotalMem += mem;
                    memLimits.addNewObjSize(mem);
                }
            }
            avgTupleSize = estTotalMem / estTuples;
            long totalTuples = memLimits.getCacheLimit();
            LOG.info("Estimated total tuples to buffer, based on " + estTuples + " tuples that took up " + estTotalMem + " bytes: " + totalTuples);
            firstTierThreshold = (int) (0.5 + totalTuples * (1f - (1f / sizeReduction)));
            secondTierThreshold = (int) (0.5 + totalTuples *  (1f / sizeReduction));
            LOG.info("Setting thresholds. Primary: " + firstTierThreshold + ". Secondary: " + secondTierThreshold);
            // The second tier should at least allow one tuple before it tries to aggregate.
            // This code retains the total number of tuples in the buffer while guaranteeing
            // the second tier has at least one tuple.
            if (secondTierThreshold == 0) {
                secondTierThreshold += 1;
                firstTierThreshold -= 1;
            }
            if (isGroupAll) {
                listSizeThreshold = Math.min(firstTierThreshold, MAX_LIST_SIZE);
            }
        }
        estimatedMemThresholds = true;
    }

    private void checkSizeReduction() throws ExecException {
        if (!mapAggDisabled()) {
            int numBeforeReduction = numRecsInProcessedMap + numRecsInRawMap;
            aggregateBothLevels(false, false);
            int numAfterReduction = numRecsInProcessedMap + numRecsInRawMap;
            LOG.info("After reduction, processed map: " + numRecsInProcessedMap + "; raw map: " + numRecsInRawMap);
            LOG.info("Observed reduction factor: from " + numBeforeReduction +
                    " to " + numAfterReduction +
                    " => " + numBeforeReduction / numAfterReduction + ".");
            if ( numBeforeReduction / numAfterReduction < minOutputReduction) {
                LOG.info("Disabling in-memory aggregation, since observed reduction is less than " + minOutputReduction);
                disableMapAgg();
            }
            sizeReduction = numBeforeReduction / numAfterReduction;
            sizeReductionChecked = true;
        }

    }
    private void disableMapAgg() throws ExecException {
        // Do not aggregate as when disableMapAgg is called aggregation is
        // called and size reduction checked
        startSpill(false);
        disableMapAgg = true;
        ALL_POPARTS.remove(this);
    }

    private boolean mapAggDisabled() {
        return disableMapAgg;
    }

    private boolean shouldAggregateFirstLevel() {
        return (numRecsInRawMap > firstTierThreshold);
    }

    private boolean shouldAggregateSecondLevel() {
        return (numRecsInProcessedMap > secondTierThreshold);
    }

    private boolean shouldSpill() {
        // is this always the same as shouldAgg?
        return shouldAggregateSecondLevel();
    }

    private void addKeySingleValToMap(Map<Object, List<Tuple>> map,
            Object key, List<Tuple> inpTuple) throws ExecException {
        List<Tuple> value = map.get(key);
        if (value == null) {
            map.put(key, inpTuple);
        } else {
            value.add(inpTuple.get(0));
        }
    }

    private void addKeyValToMap(Map<Object, List<Tuple>> map,
            Object key, Tuple inpTuple) throws ExecException {
        List<Tuple> value = map.get(key);
        if (value == null) {
            value = createNewValueList(map);
            map.put(key, value);
        }
        value.add(inpTuple);
        if (value.size() > listSizeThreshold) {
            boolean isFirst = (map == rawInputMap);
            if (LOG.isDebugEnabled()){
                LOG.debug("The cache for key " + key + " has grown too large. Aggregating " + ((isFirst) ? "first level." : "second level."));
            }
            if (isFirst) {
                // Aggregate and remove just this key to keep size in check
                aggregateRawRow(key, value);
            } else {
                aggregateSecondLevel();
            }
        }
    }

    private List<Tuple> createNewValueList(Map<Object, List<Tuple>> map) {
        List<Tuple> value;
        if (isGroupAll) {
            // Set exact array initial size to avoid array copies
            // listSizeThreshold = numRecordsToSample before estimating memory
            // thresholds and firstTierThreshold after memory estimation
            int listSize = (map == rawInputMap) ? listSizeThreshold : Math.min(secondTierThreshold + 1, MAX_LIST_SIZE);
            value = new ArrayList<Tuple>(listSize);
        } else {
            value = new ArrayList<Tuple>();
        }
        return value;
    }

    private void startSpill(boolean aggregate) throws ExecException {
        // If spillingIterator is null, we are already spilling and don't need to set up.
        if (spillingIterator != null) return;

        LOG.info("Starting spill.");
        if (aggregate) {
            aggregateBothLevels(false, false);
        }
        doSpill = true;
        spillingIterator = processedInputMap.entrySet().iterator();
    }

    private Result spillResult() throws ExecException {
        // if no more to spill, return EOP_RESULT.
        if (processedInputMap.isEmpty()) {
            spillingIterator = null;
            LOG.info("In spillResults(), processed map is empty -- done spilling.");
            return EOP_RESULT;
        } else {
            Map.Entry<Object, List<Tuple>> entry = spillingIterator.next();
            Tuple valueTuple = createValueTuple(entry.getKey(), entry.getValue());
            numRecsInProcessedMap -= entry.getValue().size();
            spillingIterator.remove();
            Result res = getOutput(entry.getKey(), valueTuple);
            return res;
        }
    }

    private void aggregateRawRow(Object key, List<Tuple> value) throws ExecException {
        numRecsInRawMap -= value.size();
        Tuple valueTuple = createValueTuple(key, value);
        Result res = getOutput(key, valueTuple);
        rawInputMap.remove(key);
        addKeyValToMap(processedInputMap, key, getAggResultTuple(res.result));
        numRecsInProcessedMap++;
    }

    /**
     * For each entry in rawInputMap, feed the list of tuples into the aggregator funcs
     * and add the results to processedInputMap. Remove the entries from rawInputMap as we go.
     * @throws ExecException
     */
    private int aggregate(Map<Object, List<Tuple>> fromMap, Map<Object, List<Tuple>> toMap, int numEntriesInTarget) throws ExecException {
        boolean srcDestDifferent = (fromMap == toMap) ? false : true;
        Iterator<Map.Entry<Object, List<Tuple>>> iter = fromMap.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry<Object, List<Tuple>> entry = iter.next();
            if (entry.getValue().size() == 1) {
                // If fromMap and toMap are same (processedInputMap), then continue without any change
                // If different (rawInputMap->processedInputMap), add directly to target and skip running through the valuePlans
                if (srcDestDifferent) {
                    iter.remove();
                    addKeySingleValToMap(toMap, entry.getKey(), entry.getValue());
                }
            } else {
                Tuple valueTuple = createValueTuple(entry.getKey(), entry.getValue());
                Result res = getOutput(entry.getKey(), valueTuple);
                if (srcDestDifferent) {
                    // Remove from src and add to destination (rawInputMap->processedInputMap)
                    iter.remove();
                    addKeyValToMap(toMap, entry.getKey(), getAggResultTuple(res.result));
                } else {
                    // Update processedInputMap in place
                    List<Tuple> value = entry.getValue();
                    if (isGroupAll) {
                        value.clear();
                    } else {
                        // Creating a new list to free more space instead of
                        // calling clear as the same key might not repeat again
                        value = createNewValueList(toMap);
                        toMap.put(entry.getKey(), value);
                    }
                    value.add(getAggResultTuple(res.result));
                }
            }
            numEntriesInTarget++;
        }
        return numEntriesInTarget;
    }

    private void aggregateBothLevels(boolean checkThresholdForFirst,
            boolean checkThresholdForSecond) throws ExecException {
        // When processed map is initially empty, just aggregate first level as
        // aggregating second level immediately would not yield anything
        boolean aggregateSecondLevel = !processedInputMap.isEmpty();
        if (!checkThresholdForFirst || shouldAggregateFirstLevel()) {
            aggregateFirstLevel();
        }
        if (aggregateSecondLevel && (!checkThresholdForSecond || shouldAggregateSecondLevel())) {
            aggregateSecondLevel();
        }
    }

    private void aggregateFirstLevel() throws ExecException {
        if (rawInputMap.isEmpty()) {
            return;
        }
        int rawTuples = numRecsInRawMap;
        int processedTuples = numRecsInProcessedMap;
        numRecsInProcessedMap = aggregate(rawInputMap, processedInputMap, numRecsInProcessedMap);
        numRecsInRawMap = 0;
        LOG.info("Aggregated " + rawTuples+ " raw tuples."
                + " Processed tuples before aggregation = " + processedTuples
                + ", after aggregation = " + numRecsInProcessedMap);
    }

    private void aggregateSecondLevel() throws ExecException {
        if (processedInputMap.isEmpty()) {
            return;
        }
        int processedTuples = numRecsInProcessedMap;
        numRecsInProcessedMap = aggregate(processedInputMap, processedInputMap, 0);
        LOG.info("Aggregated " + processedTuples + " processed tuples to " + numRecsInProcessedMap + " tuples");
    }

    private Tuple createValueTuple(Object key, List<Tuple> inpTuples) throws ExecException {
        Tuple valueTuple = mTupleFactory.newTuple(valuePlans.size() + 1);
        valueTuple.set(0, key);

        for (int i = 0; i < valuePlans.size(); i++) {
            DataBag bag = null;
            if (doContingentSpill) {
                // Don't use additional memory since we already have memory stress
                bag = new InternalCachedBag();
            } else {
                // Take 10% of memory, need fine tune later
                bag = new InternalCachedBag(1, 0.1F);
            }
            valueTuple.set(i + 1, bag);
        }
        for (Tuple t : inpTuples) {
            for (int i = 1; i < t.size(); i++) {
                DataBag bag = (DataBag) valueTuple.get(i);
                bag.add((Tuple) t.get(i));
            }
        }

        return valueTuple;
    }

    private Tuple getAggResultTuple(Object result) throws ExecException {
        try {
            return (Tuple) result;
        } catch (ClassCastException ex) {
            throw new ExecException("Intermediate Algebraic "
                    + "functions must implement EvalFunc<Tuple>");
        }
    }

    @Override
    public Tuple illustratorMarkup(Object in, Object out, int eqClassIndex) {
        // combiner optimizer does not get invoked if the plan is being executed
        // under illustrate, so POPartialAgg should not get used in that case
        throw new UnsupportedOperationException();
    }

    @Override
    public void visit(PhyPlanVisitor v) throws VisitorException {
        v.visitPartialAgg(this);
    }

    private Result getResult(ExpressionOperator op) throws ExecException {
        Result res;
        switch (op.getResultType()) {
        case DataType.BAG:
        case DataType.BOOLEAN:
        case DataType.BYTEARRAY:
        case DataType.CHARARRAY:
        case DataType.DOUBLE:
        case DataType.FLOAT:
        case DataType.INTEGER:
        case DataType.LONG:
        case DataType.BIGINTEGER:
        case DataType.BIGDECIMAL:
        case DataType.DATETIME:
        case DataType.MAP:
        case DataType.TUPLE:
            res = op.getNext(op.getResultType());
            break;
        default:
            String msg = "Invalid result type: "
                    + DataType.findType(op.getResultType());
            throw new ExecException(msg, 2270, PigException.BUG);
        }

        return res;
    }

    /**
     * Runs the provided key-value pair through the aggregator plans.
     * @param key
     * @param value
     * @return Result, containing a tuple of form (key, tupleReturnedByPlan1, tupleReturnedByPlan2, ...)
     * @throws ExecException
     */
    private Result getOutput(Object key, Tuple value) throws ExecException {
        Tuple output = mTupleFactory.newTuple(valuePlans.size() + 1);
        output.set(0, key);

        for (int i = 0; i < valuePlans.size(); i++) {
            valuePlans.get(i).attachInput(value);
            Result valRes = getResult(valueLeaves.get(i));
            if (valRes.returnStatus == POStatus.STATUS_ERR) {
                return valRes;
            }
            output.set(i + 1, valRes.result);
        }
        return new Result(POStatus.STATUS_OK, output);
    }

    @Override
    public boolean supportsMultipleInputs() {
        return false;
    }

    @Override
    public boolean supportsMultipleOutputs() {
        return false;
    }

    @Override
    public String name() {
        return getAliasString() + "Partial Agg" + "["
                + DataType.findTypeName(resultType) + "]" + mKey.toString();

    }

    public PhysicalPlan getKeyPlan() {
        return keyPlan;
    }

    public void setKeyPlan(PhysicalPlan keyPlan) {
        this.keyPlan = keyPlan;
        keyLeaf = (ExpressionOperator) keyPlan.getLeaves().get(0);
    }

    public List<PhysicalPlan> getValuePlans() {
        return valuePlans;
    }

    public void setValuePlans(List<PhysicalPlan> valuePlans) {
        this.valuePlans = valuePlans;
        valueLeaves = new ArrayList<ExpressionOperator>();
        for (PhysicalPlan plan : valuePlans) {
            valueLeaves.add((ExpressionOperator) plan.getLeaves().get(0));
        }
    }

    @Override
    public long spill() {
        if (mapAggDisabled()) {
            return 0;
        } else {
            if (doContingentSpill && !startedContingentSpill) {
                LOG.info("Spill triggered by SpillableMemoryManager, but previous spill call is still not processed. Skipping");
                return 0;
            }
            synchronized(spillLock) {
                if (rawInputMap != null) {
                    LOG.info("Spill triggered. Memory usage: " + getMemorySize()
                            + ". Raw map: keys = " + rawInputMap.size()
                            + ", tuples = "+ numRecsInRawMap
                            + ", Processed map: keys = " + processedInputMap.size()
                            + ", tuples = "+ numRecsInProcessedMap );
                }
                startedContingentSpill = false;
                doContingentSpill = true;
                if (!sizeReductionChecked || !estimatedMemThresholds) {
                    numRecordsToSample = numRecsInRawMap;
                }
                try {
                    // Block till spilling is finished. If main thread execution has not come to POPartialAgg
                    // and is still processing lower pipeline for more than 5 seconds it means
                    // jvm is stuck doing GC and will soon fail with java.lang.OutOfMemoryError: GC overhead limit exceeded
                    // So exit out of here so that SpillableMemoryManger can at least spill
                    // other Spillable bags and free up some memory for user code to be able to run
                    // and reach POPartialAgg for the aggregation/spilling of the hashmaps to happen.
                    long startTime = System.currentTimeMillis();
                    while (doContingentSpill == true) {
                        Thread.sleep(25);
                        if (!startedContingentSpill && (System.currentTimeMillis() - startTime) >= 5000) {
                            break;
                        }
                    }
                    if (doContingentSpill) {
                        LOG.info("Not blocking for spill and letting SpillableMemoryManager"
                                + " process other spillable objects as main thread has not reached here for 5 secs");
                    } else {
                        LOG.info("Finished spill for SpillableMemoryManager call");
                        return 1;
                    }
                } catch (InterruptedException e) {
                    LOG.warn("Interrupted exception while waiting for spill to finish", e);
                }
                return 0;
            }
        }
    }

    @Override
    public long getMemorySize() {
        return avgTupleSize * (numRecsInProcessedMap + numRecsInRawMap);
    }

    @Override
    public PhysicalOperator clone() throws CloneNotSupportedException {
        POPartialAgg clone = (POPartialAgg) super.clone();
        clone.setKeyPlan(keyPlan.clone());
        clone.setValuePlans(clonePlans(valuePlans));
        return clone;
    }



}
