package simpledb;

import java.util.*;

/**
 * The Aggregator operator that computes an aggregate (e.g., sum, avg, max,
 * min).  Note that we only support aggregates over a single column, grouped
 * by a single column.
 */
public class Aggregate implements DbIterator {
    private DbIterator child;
    private int afield;
    private int gfield;
    private Aggregator agg;
    private DbIterator iter;

    /**
     * Constructor.
     *
     * @param child The DbIterator that is spoonfeeding us tuples.
     * @param afield The column over which we are computing an aggregate.
     * @param gfield The column over which we are grouping the result.
     * @param agg The class that implements the aggregated value.
     */
    public Aggregate(DbIterator child, int afield, int gfield, Aggregator agg) {
        this.child = child;
        this.afield = afield;
        this.gfield = gfield;
        this.agg = agg;
    }

    public void open() 
	throws NoSuchElementException, DbException,
               TransactionAbortedException {
        child.open();

        while (true) {
            try {
                Tuple t = child.getNext();
                agg.merge(t, gfield, afield);
            } catch (NoSuchElementException e) {
                break;
            }
        }

        child.close();

        iter = agg.iterator();
        iter.open();
    }
    
    /**
     * Returns the next tuple where the first field is the field by which we are
     * grouping, and the second field is the result of computing the aggregate.
     */
    public Tuple getNext() throws TransactionAbortedException, DbException {
        return iter.getNext();
    }
    
    public void rewind() throws DbException, TransactionAbortedException {
        iter = agg.iterator();
        iter.open();
    }

    /**
     * Returns the TupleDesc of this Aggregate.
     * This is always a 2-field tuple.
     */
    public TupleDesc getTupleDesc() {
        return agg.getTupleDesc();
    }
    
    public void close() {
        iter.close();
        iter = null;
    }
}