package prefuse.data.util;

import java.util.Comparator;
import java.util.Iterator;

import prefuse.data.Table;
import prefuse.data.expression.AndPredicate;
import prefuse.data.expression.ColumnExpression;
import prefuse.data.expression.ComparisonPredicate;
import prefuse.data.expression.Expression;
import prefuse.data.expression.ExpressionAnalyzer;
import prefuse.data.expression.NotPredicate;
import prefuse.data.expression.OrPredicate;
import prefuse.data.expression.Predicate;
import prefuse.data.expression.RangePredicate;
import prefuse.data.tuple.TupleSet;
import prefuse.util.PrefuseConfig;
import prefuse.util.collections.CompositeIntIterator;
import prefuse.util.collections.IntIterator;

/**
 * Factory class that creates optimized filter iterators. When possible,
 * this factory will attempt to create an optimized query plan by using
 * available indexes, in many incrasing performance by only visiting
 * the tuples which will pass the filter condition.
 * 
 * @author <a href="http://jheer.org">jeffrey heer</a>
 */
public class FilterIteratorFactory {

    private static final int OPTIMIZATION_THRESHOLD
        = PrefuseConfig.getInt("data.filter.optimizeThreshold");
    
    // we can stash our query plan generation and optimization here to deal 
    // with it all in one spot, and keep the rest of the classes clean
    
    /**
     * Get a filtered iterator over the tuples in the given set,
     * filtered by the given predicate.
     * @param ts the TupleSet to iterate over
     * @param p the filter predicate
     * @return a filtered iterator over the tuples
     */
    public static Iterator tuples(TupleSet ts, Predicate p) {
        // no predicate means no filtering
        if ( p == null )
            return ts.tuples();
        
        // attempt to generate an optimized query plan
        Iterator iter = null;
        if ( ts instanceof Table ) {
            Table t = (Table)ts;
            IntIterator ii = getOptimizedIterator(t,p);
            if ( ii != null )
                iter = t.tuples(ii);
        }
        
        // optimization fails, scan the entire table
        if ( iter == null ) {
            iter = new FilterIterator(ts.tuples(), p);
        }
        
        return iter;
    }
    
    /**
     * Get a filtered iterator over the rows in the given table,
     * filtered by the given predicate.
     * @param t the Table to iterate over
     * @param p the filter predicate
     * @return a filtered iterator over the table rows
     */
    public static IntIterator rows(Table t, Predicate p) {
        // attempt to generate an optimized query plan
        IntIterator iter = null;
        iter = getOptimizedIterator(t, p);
        
        // optimization fails, scan the entire table
        if ( iter == null ) {
            iter = new FilterRowIterator(t.rows(), t, p);
        }
        return iter;
    }
    
    /**
     * Get an optimized iterator over the rows of a table, if possible.
     * @param t the Table to iterator over
     * @param p the filter predicate
     * @return an optimized iterator, or null if no optimization was found
     */
    protected static IntIterator getOptimizedIterator(Table t, Predicate p) {
        if ( t.getRowCount() < OPTIMIZATION_THRESHOLD ) 
            return null; // avoid overhead for small tables
        
        if ( p instanceof ColumnExpression ) {
            // try to optimize a boolean column
            return getColumnIterator(t,
                    ((ColumnExpression)p).getColumnName(), true);
        }
        else if ( p instanceof NotPredicate )
        {
            // try to optimize the negation a boolean column
            Predicate pp = ((NotPredicate)p).getPredicate();
            if ( pp instanceof ColumnExpression ) {
                return getColumnIterator(t,
                        ((ColumnExpression)pp).getColumnName(), false);
            }
        }
        else if ( p instanceof AndPredicate )
        {
            // try to optimize an and clause
            return getAndIterator(t, (AndPredicate)p);
        }
        else if ( p instanceof OrPredicate )
        {
            // try to optimize an or clause
            return getOrIterator(t, (OrPredicate)p);
        }
        else if ( p instanceof ComparisonPredicate )
        {
            // try to optimize a comparison (=, !=, <, > ,etc)
            return getComparisonIterator(t,(ComparisonPredicate)p);
        }
        else if ( p instanceof RangePredicate )
        {
            // try to optimize a bounded range of values
            return getRangeIterator(t, (RangePredicate)p); 
        }
        
        return null;
    }
    
    protected static IntIterator getColumnIterator(
            Table t, String field, boolean val)
    {
        if ( t.getColumnType(field) != boolean.class )
            return null; // only works for boolean-valued columns
        
        Index index = t.getIndex(field);
        if ( index == null ) {
            return null;
        } else {
            return index.rows(val);
        }
    }
    
    protected static IntIterator getOrIterator(Table t, OrPredicate op) {
        int size = op.size();
        if ( size > 1 ) {
            // if all subclauses can be optimized, we can optimize the query
            IntIterator[] rows = new IntIterator[size];
            for ( int i=0; i<rows.length; ++i ) {
                rows[i] = getOptimizedIterator(t, op.get(i));
                
                // all clauses must be optimized to avoid linear scan
                if ( rows[i] == null ) return null;
            }
            // group iterators, and filter for uniqueness
            return new UniqueRowIterator(new CompositeIntIterator(rows));
        } else if ( size == 1 ) {
            // only one clause, optimize for that
            return getOptimizedIterator(t, op.get(0));
        } else {
            // no woman, no cry
            return null;
        }
    }
    
    protected static IntIterator getAndIterator(Table t, AndPredicate ap) {
        // possible TODO: add scoring to select best optimized iterator
        // for now just work from the end backwards and take the first
        // optimized iterator we find
        IntIterator rows = null;
        Predicate clause = null;
        for ( int i=ap.size(); --i >= 0; ) {
            clause = ap.get(i);
            if ( (rows=getOptimizedIterator(t,clause)) != null )
                break;
        }
        
        // exit if we didn't optimize
        if ( rows == null ) return null;
        
        // if only one clause, no extras needed
        if ( ap.size() == 1 ) return rows;
        
        // otherwise get optimized source, run through other clauses
        return new FilterRowIterator(rows, t, ap.getSubPredicate(clause));
    }
    
    protected static IntIterator getComparisonIterator(Table t, 
                                           ComparisonPredicate cp)
    {
        Expression l = cp.getLeftExpression();
        Expression r = cp.getRightExpression();
        int operation = cp.getOperation();
        
        // not equals operations aren't handled by the index
        if ( operation == ComparisonPredicate.NEQ )
            return null;
        
        ColumnExpression col;
        Expression lit;
        
        // make sure columns are of the right type
        if (l instanceof ColumnExpression && 
                !ExpressionAnalyzer.hasDependency(r))
        {
            col = (ColumnExpression)l;
            lit = r;
        } else if (r instanceof ColumnExpression &&
                !ExpressionAnalyzer.hasDependency(l))
        {
            col = (ColumnExpression)r;
            lit = l;
        } else {
            return null;
        }
        
        // if table has index of the right type, use it
        Comparator cmp = cp.getComparator();
        Index index = t.getIndex(col.getColumnName());
        
        if ( index == null || !cmp.equals(index.getComparator()) )
            return null;
        
        Class ltype = lit.getClass();
        if ( ltype == int.class ) {
            int val = lit.getInt(null); // literal value, so null is safe
            switch ( operation ) {
            case ComparisonPredicate.LT:
                return index.rows(Integer.MIN_VALUE, val, Index.TYPE_AIE);
            case ComparisonPredicate.GT:
                return index.rows(val, Integer.MAX_VALUE, Index.TYPE_AEI);
            case ComparisonPredicate.EQ:
                return index.rows(val, val, Index.TYPE_AII);
            case ComparisonPredicate.LTEQ:
                return index.rows(Integer.MIN_VALUE, val, Index.TYPE_AII);
            case ComparisonPredicate.GTEQ:
                return index.rows(val, Integer.MAX_VALUE, Index.TYPE_AII);
            default:
                throw new IllegalStateException(); // should never occur
            }
        } else if ( ltype == long.class ) {
            long val = lit.getLong(null); // literal value, so null is safe
            switch ( operation ) {
            case ComparisonPredicate.LT:
                return index.rows(Long.MIN_VALUE, val, Index.TYPE_AIE);
            case ComparisonPredicate.GT:
                return index.rows(val, Long.MAX_VALUE, Index.TYPE_AEI);
            case ComparisonPredicate.EQ:
                return index.rows(val, val, Index.TYPE_AII);
            case ComparisonPredicate.LTEQ:
                return index.rows(Long.MIN_VALUE, val, Index.TYPE_AII);
            case ComparisonPredicate.GTEQ:
                return index.rows(val, Long.MAX_VALUE, Index.TYPE_AII);
            default:
                throw new IllegalStateException(); // should never occur
            }
        } else if ( ltype == float.class ) {
            float val = lit.getFloat(null); // literal value, so null is safe
            switch ( operation ) {
            case ComparisonPredicate.LT:
                return index.rows(Float.MIN_VALUE, val, Index.TYPE_AIE);
            case ComparisonPredicate.GT:
                return index.rows(val, Float.MAX_VALUE, Index.TYPE_AEI);
            case ComparisonPredicate.EQ:
                return index.rows(val, val, Index.TYPE_AII);
            case ComparisonPredicate.LTEQ:
                return index.rows(Float.MIN_VALUE, val, Index.TYPE_AII);
            case ComparisonPredicate.GTEQ:
                return index.rows(val, Float.MAX_VALUE, Index.TYPE_AII);
            default:
                throw new IllegalStateException(); // should never occur
            }
        } else if ( ltype == double.class ) {
            double val = lit.getDouble(null); // literal value, so null is safe
            switch ( operation ) {
            case ComparisonPredicate.LT:
                return index.rows(Double.MIN_VALUE, val, Index.TYPE_AIE);
            case ComparisonPredicate.GT:
                return index.rows(val, Double.MAX_VALUE, Index.TYPE_AEI);
            case ComparisonPredicate.EQ:
                return index.rows(val, val, Index.TYPE_AII);
            case ComparisonPredicate.LTEQ:
                return index.rows(Double.MIN_VALUE, val, Index.TYPE_AII);
            case ComparisonPredicate.GTEQ:
                return index.rows(val, Double.MAX_VALUE, Index.TYPE_AII);
            default:
                throw new IllegalStateException(); // should never occur
            }
        } else {
            Object val = lit.get(null); // literal value, so null is safe
            switch ( operation ) {
            case ComparisonPredicate.LT:
                return index.rows(null, val, Index.TYPE_AIE);
            case ComparisonPredicate.GT:
                return index.rows(val, null, Index.TYPE_AEI);
            case ComparisonPredicate.EQ:
                return index.rows(val, val, Index.TYPE_AII);
            case ComparisonPredicate.LTEQ:
                return index.rows(null, val, Index.TYPE_AII);
            case ComparisonPredicate.GTEQ:
                return index.rows(val, null, Index.TYPE_AII);
            default:
                throw new IllegalStateException(); // should never occur
            }
        }        
    }
    
    protected static IntIterator getRangeIterator(Table t, RangePredicate rp) {
        ColumnExpression col;
        Expression l, r;
        
        // make sure columns are of the right type
        if ( !(rp.getMiddleExpression() instanceof ColumnExpression) ||
                ExpressionAnalyzer.hasDependency(rp.getLeftExpression()) ||
                ExpressionAnalyzer.hasDependency(rp.getRightExpression()) )
        {
            return null;
        }
        
        // assign variables
        col = (ColumnExpression)rp.getMiddleExpression();
        l = rp.getLeftExpression();
        r = rp.getRightExpression();
        
        // if table has index of the right type, use it
        Comparator cmp = rp.getComparator();
        Index index = t.getIndex(col.getColumnName());
        
        if ( index == null || !cmp.equals(index.getComparator()) )
            return null;
        
        int operation = rp.getOperation();
        Class ltype = t.getColumnType(col.getColumnName());
        
        // TODO safety check literal types
        
        // get the index type
        int indexType;
        switch ( operation ) {
        case RangePredicate.IN_IN:
            indexType = Index.TYPE_AII;
            break;
        case RangePredicate.IN_EX:
            indexType = Index.TYPE_AIE;
            break;
        case RangePredicate.EX_IN:
            indexType = Index.TYPE_AEI;
            break;
        case RangePredicate.EX_EX:
            indexType = Index.TYPE_AEE;
            break;
        default:
            throw new IllegalStateException(); // should never occur
        }
        
        // get the indexed rows
        if ( ltype == int.class ) {
            return index.rows(l.getInt(null), r.getInt(null), indexType);
        } else if ( ltype == long.class ) {
            return index.rows(l.getLong(null), r.getLong(null), indexType);
        } else if ( ltype == float.class ) {
            return index.rows(l.getFloat(null), r.getFloat(null), indexType);
        } else if ( ltype == double.class ) {
            return index.rows(l.getDouble(null), r.getDouble(null), indexType);
        } else {
            return index.rows(l.get(null), r.get(null), indexType);
        }
    }
    
} // end of class FilterIteratorFactory