package simpledb;
import java.util.*;

/**
 * The Join operator implements the relational join operation.
 */
public class Join implements DbIterator {
    private JoinPredicate pred;
    private DbIterator child1;
    private DbIterator child2;
    private Tuple t1;
    
    /**
     * Constructor.  Accepts to children to join and the predicate
     * to join them on
     *
     * @param p The predicate to use to join the children
     * @param child1 Iterator for the left(outer) relation to join
     * @param child2 Iterator for the right(inner) relation to join
     */
    public Join(JoinPredicate p, DbIterator child1, DbIterator child2) {
        this.pred = p;
        this.child1 = child1;
        this.child2 = child2;
    }
    
    public TupleDesc getTupleDesc() {
        return TupleDesc.combine(child1.getTupleDesc(),
                                 child2.getTupleDesc());
    }
    
    public void open() 
	throws DbException, NoSuchElementException, TransactionAbortedException {
        child1.open();
        child2.open();
    }
    
    public void close() {
        child1.close();
        child2.close();
    }
    
    public void rewind() throws DbException, TransactionAbortedException {
        child1.rewind();
        child2.rewind();
    }
    
    /**
     * Returns the next tuple generated by the join.
     * Logically, this is the next tuple in r1 cross r2 that satisfies the join
     * predicate.  There are many possible implementations; the simplest is a
     * nested loops join.
     * <p>
     * Note that the tuples returned from this particular implpementation of
     * Join are simply the concatenation of joining tuples from the left and
     * right relation. Therefore, there will be two copies of the join attribute
     * in the results.  (Removing such duplicate columns can be done with an
     * additional projection operator if needed.)
     * <p>
     * For example, if one tuple is {1,2,3} and the other tuple is {1,5,6},
     * joined on equality of the first column, then this returns {1,2,3,1,5,6}.
     *
     * @return The next matching tuple.
     * @see JoinPredicate#filter
     */
    public Tuple getNext() 
	throws NoSuchElementException, TransactionAbortedException, DbException {
        Tuple t2;
        
        if (t1 == null) {
            t1 = child1.getNext();
        }

        /*
         * Nested loops join.
         *
         * t1 is the tuple from the outer table we're currently
         * joining against the tuples (as t2) of the other table. We
         * iterate over the inner table until we run out of elements,
         * then move on to the next element in the outer table.
         */
        while (true) {
            try {
                t2 = child2.getNext();
            } catch (NoSuchElementException e) {
                try {
                    t1 = child1.getNext();
                } catch (NoSuchElementException e2) {
                    // We've exhausted the outer table, so we're
                    // done. Re-throw the exception.
                    throw e2;
                }

                child2.rewind();
                t2 = child2.getNext();
            }

            if (pred.filter(t1, t2)) {
                return Tuple.combine(t1, t2);
            }
        }
    }
}