import com.sleepycat.je.Database;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.je.LockMode;  
import com.sleepycat.je.OperationStatus; 

import java.io.*;
import java.util.Vector;
import java.util.Hashtable;

public class HashJoinOperator extends JoinOperator {
    static boolean canBeUsed(Operator left, Operator right, Predicate jp) {
        return true;
    }

    void init() {
        /* Don't need to do anything special. */
        super.init();

        hashtable = new Hashtable();
    }

    HashJoinOperator(Operator l, Operator r, Predicate jp) {
        super(l, r, jp);
    }

    /* Pretty print for the query plan. */
    void print(int num_tabs) {
        for(int i = 0; i < num_tabs; i++) 
            System.out.print("	");
        System.out.println("Hash Join operator with predicate " + jp);
        leftOp.print(num_tabs+1);
        rightOp.print(num_tabs+1);
    }

    boolean firstCall = true;
    Hashtable hashtable = null;

    Tuple leftTuple = null;
    Vector<Tuple> rightTuplesMatching = null;
    int rightTuplesReturned = -1;

    Tuple get_next() {
        /* If it is the first call, then we should built the hash table on the right table. This is usually done here instead
           of the init(). The convention is not to do any data manipulation during init()'s. */
        if(firstCall) {
            Tuple rightTuple = null;
            while((rightTuple = rightOp.get_next()) != null) {
                /* Insert into the hash table. */
                Object key = jp.rhs().evaluate(rightTuple);
                Vector v = (Vector) hashtable.get(key);
                if(v == null) {
                    v = new Vector<Tuple>();
                    v.add(rightTuple);
                    hashtable.put(key, v);
                } else {
                    v.add(rightTuple);
                }
            }
            firstCall = false;
        }

        /* If there are still tuples remaining corresponding to the current left tuple, create and return an intermediate tuple.
           Else get the next left tuple. */
        while(true) {
            if(leftTuple == null) { 
                if((leftTuple = leftOp.get_next()) == null)
                    return null;

                Object key = jp.lhs().evaluate(leftTuple);
                rightTuplesMatching = (Vector) hashtable.get(key);

                if(rightTuplesMatching != null) {
                    rightTuplesReturned = 0;
                    /* Note that, after this, the loop will get executed again, and this time since leftTuple != null, the "else" will be executed. */
                } else { 
                    leftTuple = null;
                }
            } else {
                if(rightTuplesReturned < rightTuplesMatching.size()) {
                    Tuple ret = new IntermediateTuple(leftTuple, rightTuplesMatching.get(rightTuplesReturned));
                    rightTuplesReturned++;
                    return ret;
                } else {
                    /* We have exhausted them. */
                    leftTuple = null;
                }
            }
        }
    }

    void close() {
        /* Don't need to do anything special. */
        super.close();
    }
}