package edu.caltech.nanodb.plannodes;


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

import edu.caltech.nanodb.expressions.OrderByExpression;
import edu.caltech.nanodb.expressions.TupleLiteral;
import edu.caltech.nanodb.relations.Tuple;


/**
 * <p>
 * This plan-node materializes the results of a child plan-node in memory.
 * The tuples of the child plan-node are fetched on-demand (not all at once
 * at the start of plan-node execution), and are cached within this plan-node.
 * If the child plan produces disk-backed tuples, this plan-node caches a copy
 * of the tuple so that the disk-backed tuple can be unpinned.
 * </p>
 * <p>
 * Note that a more typical implementation of a "materialize" node would have
 * a maximum in-memory footprint, and would store tuples to a temporary disk
 * file when this maximum memory size is reached.  However, the memory
 * allocation for in-memory tuples is not managed by the Buffer Manager, so it
 * really isn't possible for the materialize node to provide this kind of
 * functionality.
 * </p>
 */
public class MaterializeNode extends PlanNode {

    /**
     * This collection holds the tuples that have been generated by the child
     * plan so far.
     */
    private ArrayList<Tuple> tuples;


    /**
     * This field stores the index of the "current tuple" as the materialized
     * results are traversed.
     */
    private int currentTupleIndex;


    /** This field stores the index of the tuple when a position is marked. */
    private int markedTupleIndex;



    /**
     * When this flag is true, the child plan-node has finished generating
     * all of its results.
     */
    private boolean childNodeFinished;



    public MaterializeNode(PlanNode leftChild) {
        super(leftChild);
    }


    @Override
    public List<OrderByExpression> resultsOrderedBy() {
        return leftChild.resultsOrderedBy();
    }


    @Override
    public boolean supportsMarking() {
        return true;
    }


    @Override
    public void prepare() {
        leftChild.prepare();

        schema = leftChild.getSchema();
        stats = leftChild.getStats();
        cost = leftChild.getCost();

        tuples = new ArrayList<>();
        currentTupleIndex = -1;
        markedTupleIndex = -1;
        childNodeFinished = false;
    }


    @Override
    public Tuple getNextTuple() {
        Tuple tup = null;

        assert currentTupleIndex >= -1;
        assert currentTupleIndex <= tuples.size();

        if (currentTupleIndex + 1 < tuples.size()) {
            // Moving forward the "current tuple" index will stay within the
            // tuples we have so far.
            currentTupleIndex++;
            tup = tuples.get(currentTupleIndex);
        }
        else {
            // Moving forward the "current tuple" index will move beyond the
            // tuples we have so far.  Need to see if we have consumed all
            // child tuples yet.

            if (!childNodeFinished) {
                // Try to fetch another tuple from the child plan-node, if
                // there is one.
                tup = leftChild.getNextTuple();
                if (tup != null) {
                    if (tup.isDiskBacked()) {
                        // Make an in-memory version of the tuple we can cache.
                        Tuple copy = new TupleLiteral(tup);
                        tup.unpin();
                        tup = copy;
                    }

                    tuples.add(tup);
                    currentTupleIndex++;
                }
                else {
                    // The child has no more tuples.
                    childNodeFinished = true;
                }

                assert currentTupleIndex <= tuples.size();
            }
        }

        return tup;
    }


    @Override
    public void markCurrentPosition() {
        markedTupleIndex = currentTupleIndex;
    }

    @Override
    public void resetToLastMark() {
        currentTupleIndex = markedTupleIndex;
    }

    @Override
    public void cleanUp() {
        leftChild.cleanUp();

        tuples = null;
        currentTupleIndex = -1;
    }

    @Override
    public String toString() {
        return "Materialize";
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof MaterializeNode) {
            MaterializeNode other = (MaterializeNode) obj;
            return leftChild.equals(other.leftChild);
        }

        return false;
    }

    @Override
    public int hashCode() {
        return leftChild.hashCode();
    }
}