package simpledb;
import java.util.*;

/**
 * INLJoin is an index nested loops implementation for SimpleDB.
 * 
 */
public class INLJoin implements DbIterator {
    private DbIterator child1;
    private IndexDbIterator child2;
    private Tuple t1;
    private int field1;

    /**
     * Constructor.  Accepts children to join and the predicate
     * to join them on
     *
     * @param child1 Iterator for the left(outer) relation to join
     * @param child2 Iterator for the right(inner) relation to join
     *        child2 has an index (e.g., hash or B tree) on its key field
     *               and predicate has already been included in it.
     * @param field1 child1's field to compare with child2
     */

    public INLJoin(DbIterator child1, int field1, IndexDbIterator child2) {
        this.child1 = child1;
        this.field1 = field1;
        this.child2 = child2;
    }

    public TupleDesc getTupleDesc() {
        return TupleDesc.combine(child1.getTupleDesc(),
                                 child2.getTupleDesc());
    }

    public void open()
	throws DbException, NoSuchElementException, TransactionAbortedException {
        child1.open();
        child2.open();
        t1 = null;
    }

    public void close() {
        child1.close();
        child2.close();
    }

    public void rewind() throws DbException, TransactionAbortedException {
        child1.rewind();
        t1 = null;
    }

    public Tuple getNext() 
	throws NoSuchElementException, TransactionAbortedException, DbException {
        Tuple t2;

        // Get the first element if necessary
        if (t1 == null) {
            t1 = child1.getNext();
            child2.rewind(new IndexPredicate(Predicate.Op.EQUALS,
                                             t1.getField(field1)));
        }

        // Find a matching tuple to join. This might take several
        // attempts because tuples in the outer table might have no
        // matches in the inner index, i.e. child2 will immediately
        // fail.
        t2 = null;
        while (t2 == null) {
            try {
                t2 = child2.getNext();
            } catch (NoSuchElementException e) {
                // Inner table has been exhausted; try to find a new
                // tuple from the outer table
                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(new IndexPredicate(Predicate.Op.EQUALS,
                                                 t1.getField(field1)));
            }
        }
        
        return Tuple.combine(t1, t2);
    }
}