package edu.washington.escience.myriad.operator;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import edu.washington.escience.myriad.DbException;
import edu.washington.escience.myriad.Schema;
import edu.washington.escience.myriad.Type;
import edu.washington.escience.myriad.table._TupleBatch;

public class DupElim extends Operator {

  /** Required for Java serialization. */
  private static final long serialVersionUID = 1L;

  private class IndexedTuple {
    int index;
    _TupleBatch tb;

    public IndexedTuple(final _TupleBatch tb, final int index) {
      this.tb = tb;
      this.index = index;
    }

    public boolean compareField(final IndexedTuple another, final int colIndx) {
      final Type type = this.tb.inputSchema().getFieldType(colIndx);
      final int rowIndx1 = this.index;
      final int rowIndx2 = another.index;
      // System.out.println(rowIndx1 + " " + rowIndx2 + " " + colIndx + " " + type);
      if (type.equals(Type.INT_TYPE)) {
        return this.tb.getInt(colIndx, rowIndx1) == another.tb.getInt(colIndx, rowIndx2);
      }
      if (type.equals(Type.DOUBLE_TYPE)) {
        return this.tb.getDouble(colIndx, rowIndx1) == another.tb.getDouble(colIndx, rowIndx2);
      }
      if (type.equals(Type.STRING_TYPE)) {
        return this.tb.getString(colIndx, rowIndx1).equals(another.tb.getString(colIndx, rowIndx2));
      }
      if (type.equals(Type.FLOAT_TYPE)) {
        return this.tb.getFloat(colIndx, rowIndx1) == another.tb.getFloat(colIndx, rowIndx2);
      }
      if (type.equals(Type.BOOLEAN_TYPE)) {
        return this.tb.getBoolean(colIndx, rowIndx1) == another.tb.getBoolean(colIndx, rowIndx2);
      }
      if (type.equals(Type.LONG_TYPE)) {
        return this.tb.getLong(colIndx, rowIndx1) == another.tb.getLong(colIndx, rowIndx2);
      }
      return false;
    }

    @Override
    public boolean equals(final Object o) {
      if (!(o instanceof IndexedTuple)) {
        return false;
      }
      final IndexedTuple another = (IndexedTuple) o;
      if (!(this.tb.inputSchema().equals(another.tb.inputSchema()))) {
        return false;
      }
      for (int i = 0; i < this.tb.inputSchema().numFields(); ++i) {
        if (!compareField(another, i)) {
          return false;
        }
      }
      return true;
    }

    @Override
    public int hashCode() {
      return this.tb.hashCode(this.index);
    }
  }

  Operator child;
  Schema outputSchema;
  HashMap<Integer, List<IndexedTuple>> uniqueTuples;

  public DupElim(final Schema outputSchema, final Operator child) {
    this.outputSchema = outputSchema;
    this.child = child;
    this.uniqueTuples = new HashMap<Integer, List<IndexedTuple>>();
  }

  @Override
  protected _TupleBatch fetchNext() throws DbException {
    if (child.hasNext()) {
      final _TupleBatch tb = child.next();

      for (int i = 0; i < tb.numInputTuples(); ++i) {
        final IndexedTuple cntTuple = new IndexedTuple(tb, i);
        final int cntHashCode = cntTuple.hashCode();
        // might need to check invalid | change to use outputTuples later
        if (uniqueTuples.get(cntHashCode) == null) {
          uniqueTuples.put(cntHashCode, new ArrayList<IndexedTuple>());
        }
        final List<IndexedTuple> tupleList = uniqueTuples.get(cntHashCode);
        boolean unique = true;
        for (int j = 0; j < tupleList.size(); ++j) {
          final IndexedTuple oldTuple = tupleList.get(j);
          if (cntTuple.equals(oldTuple)) {
            unique = false;
            break;
          }
        }
        System.out.println(i + " " + unique);
        if (unique) {
          tupleList.add(cntTuple);
        } else {
          tb.remove(i);
        }
      }
      return tb;
    }
    return null;
  }

  @Override
  public Operator[] getChildren() {
    return new Operator[] { this.child };
  }

  @Override
  public Schema getSchema() {
    return this.outputSchema;
  }

  @Override
  public void open() throws DbException {
    if (child != null) {
      // for (Operator child : children)
      child.open();
    }
    super.open();
  }

  @Override
  public void setChildren(final Operator[] children) {
    this.child = children[0];
  }

}

@@ -745,21 +745,16 @@
  type = 
-this.
 tb.input
@@ -812,21 +812,16 @@
 Indx1 = 
-this.
 index;%0A 
@@ -997,29 +997,24 @@
      return 
-this.
 tb.getInt(co
@@ -1133,29 +1133,24 @@
      return 
-this.
 tb.getDouble
@@ -1275,29 +1275,24 @@
      return 
-this.
 tb.getString
@@ -1421,29 +1421,24 @@
      return 
-this.
 tb.getFloat(
@@ -1562,29 +1562,24 @@
      return 
-this.
 tb.getBoolea
@@ -1708,21 +1708,16 @@
  return 
-this.
 tb.getLo
@@ -2010,21 +2010,16 @@
   if (!(
-this.
 tb.input
@@ -2120,21 +2120,16 @@
  0; i %3C 
-this.
 tb.input
@@ -2321,21 +2321,16 @@
  return 
-this.
 tb.hashC
@@ -2333,21 +2333,16 @@
 ashCode(
-this.
 index);%0A
@@ -2374,31 +2374,8 @@
 ld;%0A
-  Schema outputSchema;%0A
   Ha
@@ -2451,90 +2451,25 @@
 nal 
-Schema outputSchema, final Operator child) %7B%0A    this.outputSchema = outputSchema;
+Operator child) %7B
 %0A   
@@ -2493,21 +2493,16 @@
 ld;%0A    
-this.
 uniqueTu
@@ -3689,21 +3689,16 @@
 tor%5B%5D %7B 
-this.
 child %7D;
@@ -3760,25 +3760,25 @@
 urn 
-this.outpu
+child.ge
 tSchema
+()
 ;%0A  
@@ -4012,37 +4012,32 @@
 children) %7B%0A    
-this.
 child = children

remove explicit Schema parameter in constructor because the schema can be infered from the child