package edu.berkeley.cs186.database.query;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

import edu.berkeley.cs186.database.Database;
import edu.berkeley.cs186.database.DatabaseException;
import edu.berkeley.cs186.database.databox.DataBox;
import edu.berkeley.cs186.database.io.Page;
import edu.berkeley.cs186.database.table.Record;
import edu.berkeley.cs186.database.table.RecordIterator;
import edu.berkeley.cs186.database.table.Schema;

import javax.xml.crypto.Data;

public class SortMergeOperator extends JoinOperator {

  public SortMergeOperator(QueryOperator leftSource,
           QueryOperator rightSource,
           String leftColumnName,
           String rightColumnName,
           Database.Transaction transaction) throws QueryPlanException, DatabaseException {
    super(leftSource, rightSource, leftColumnName, rightColumnName, transaction, JoinType.SORTMERGE);

  }

  public Iterator<Record> iterator() throws QueryPlanException, DatabaseException {
    return new SortMergeOperator.SortMergeIterator();
  }

  /**
   * An implementation of Iterator that provides an iterator interface for this operator.
   *
   * Before proceeding, you should read and understand SNLJOperator.java
   *    You can find it in the same directory as this file.
   *
   * Word of advice: try to decompose the problem into distinguishable sub-problems.
   *    This means you'll probably want to add more methods than those given (Once again,
   *    SNLJOperator.java might be a useful reference).
   *
   * Also, see discussion slides for week 7.
   */
  private class SortMergeIterator extends JoinIterator {
    /**
    * Some member variables are provided for guidance, but there are many possible solutions.
    * You should implement the solution that's best for you, using any member variables you need.
    * You're free to use these member variables, but you're not obligated to.
    */

     private String leftTableName;
     private String rightTableName;
     private RecordIterator leftIterator;
     private RecordIterator rightIterator;
     private Record leftRecord;
     private Record nextRecord;
     private Record rightRecord;
     private int marked;
     private Comparator<Record> comparator=new LR_RecordComparator();
    public SortMergeIterator() throws QueryPlanException, DatabaseException {
      super();
      SortOperator left=new SortOperator(getTransaction(),getLeftTableName(),new LeftRecordComparator());
      SortOperator right= new SortOperator(getTransaction(),getLeftTableName(),new RightRecordComparator());
      leftIterator=getRecordIterator(left.sort());
      rightIterator=getRecordIterator(right.sort());
      marked=2;
    }

    /**
     * Checks if there are more record(s) to yield
     *
     * @return true if this iterator has another record to yield, otherwise false
     */
    public boolean hasNext() {
        if(nextRecord != null){return true;}
        if(!((leftRecord !=null || leftIterator.hasNext()))){return false;}
        Record left,right;
        if(leftRecord==null){
            left=leftIterator.next();
        }
        else{
            left=leftRecord;
        }
        if(marked==0){
            rightIterator.mark();
            marked=1;
        }
        if(!rightIterator.hasNext()){
            if(leftIterator.hasNext()){
                leftRecord=leftIterator.next();
            }
            else{
                return false;
            }
            rightIterator.reset();
            return hasNext();
        }
        else{
            right=rightIterator.next();
        }
        if(marked==2){
            marked=1;
            rightIterator.mark();
        }
        if(leftRecord==null){
            leftRecord=left;
        }
        while (comparator.compare(left,right)!=0){
            if(comparator.compare(left,right)>=0){
                if(rightIterator.hasNext()){
                    right=rightIterator.next();
                    marked=0;
                }
                else{return false;}
            }
            else{
                if(leftIterator.hasNext()){
                    left=leftIterator.next();
                    leftRecord=left;
                    rightIterator.reset();
                    right=rightIterator.next();
                }
                else{return false;}
            }
        }
        ArrayList<DataBox> leftvals=new ArrayList<>(left.getValues());
        ArrayList<DataBox> rightvals=new ArrayList<>(right.getValues());
        leftvals.addAll(rightvals);
        nextRecord=new Record(leftvals);
        return true;
    }


    /**
     * Yields the next record of this iterator.
     *
     * @return the next Record
     * @throws NoSuchElementException if there are no more Records to yield
     */
    public Record next() {
      if (nextRecord != null) {
        Record ret = nextRecord;
        nextRecord = null;
        return ret;
      }
      throw new NoSuchElementException();
    }

    public void remove() {
      throw new UnsupportedOperationException();
    }


    private class LeftRecordComparator implements Comparator<Record> {
      public int compare(Record o1, Record o2) {
        return o1.getValues().get(SortMergeOperator.this.getLeftColumnIndex()).compareTo(
                o2.getValues().get(SortMergeOperator.this.getLeftColumnIndex()));
      }
    }

    private class RightRecordComparator implements Comparator<Record> {
      public int compare(Record o1, Record o2) {
        return o1.getValues().get(SortMergeOperator.this.getRightColumnIndex()).compareTo(
                o2.getValues().get(SortMergeOperator.this.getRightColumnIndex()));
      }
    }

    /**
    * Left-Right Record comparator
    * o1 : leftRecord
    * o2: rightRecord
    */
    private class LR_RecordComparator implements Comparator<Record> {
      public int compare(Record o1, Record o2) {
        return o1.getValues().get(SortMergeOperator.this.getLeftColumnIndex()).compareTo(
                o2.getValues().get(SortMergeOperator.this.getRightColumnIndex()));
      }
    }
  }
}