/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.hyracks.dataflow.std.join;

import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hyracks.api.comm.IFrameTupleAccessor;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
import org.apache.hyracks.dataflow.std.buffermanager.DeallocatableFramePool;
import org.apache.hyracks.dataflow.std.buffermanager.IDeallocatableFramePool;
import org.apache.hyracks.dataflow.std.buffermanager.IDeletableTupleBufferManager;
import org.apache.hyracks.dataflow.std.buffermanager.ITupleAccessor;
import org.apache.hyracks.dataflow.std.buffermanager.ITuplePointerAccessor;
import org.apache.hyracks.dataflow.std.buffermanager.TupleAccessor;
import org.apache.hyracks.dataflow.std.buffermanager.VariableDeletableTupleMemoryManager;
import org.apache.hyracks.dataflow.std.structures.RunFilePointer;
import org.apache.hyracks.dataflow.std.structures.TuplePointer;

/**
 * Merge Joiner takes two sorted streams of input and joins.
 * The two sorted streams must be in a logical order and the comparator must
 * support keeping that order so the join will work.
 * The left stream will spill to disk when memory is full.
 * The right stream spills to memory and pause when memory is full.
 */
public class MergeJoiner extends AbstractMergeJoiner {

    private static final Logger LOGGER = Logger.getLogger(MergeJoiner.class.getName());

    private final IDeallocatableFramePool framePool;
    private final IDeletableTupleBufferManager bufferManager;
    private final ITuplePointerAccessor memoryAccessor;
    private final LinkedList<TuplePointer> memoryBuffer = new LinkedList<>();

    private int leftStreamIndex;
    private final RunFileStream runFileStreamOld;
    private final RunFileStream runFileStream;
    private ITupleAccessor tmpAccessor;
    private final RunFilePointer runFilePointer;

    private final IMergeJoinChecker mjc;

    private long joinComparisonCount = 0;
    private long joinResultCount = 0;
    private long spillFileCount = 0;
    private long spillWriteCount = 0;
    private long spillReadCount = 0;
    private long spillCount = 0;

    public MergeJoiner(IHyracksTaskContext ctx, int memorySize, int partition, MergeStatus status, MergeJoinLocks locks,
            IMergeJoinChecker mjc, RecordDescriptor leftRd, RecordDescriptor rightRd) throws HyracksDataException {
        super(ctx, partition, status, locks, leftRd, rightRd);
        this.mjc = mjc;
        tmpAccessor = new TupleAccessor(leftRd);
        runFilePointer = new RunFilePointer();

        // Memory (right buffer)
        if (memorySize < 1) {
            throw new HyracksDataException(
                    "MergeJoiner does not have enough memory (needs > 0, got " + memorySize + ").");
        }
        framePool = new DeallocatableFramePool(ctx, (memorySize) * ctx.getInitialFrameSize());
        bufferManager = new VariableDeletableTupleMemoryManager(framePool, rightRd);
        memoryAccessor = bufferManager.createTuplePointerAccessor();

        // Run File and frame cache (left buffer)
        leftStreamIndex = TupleAccessor.UNSET;
        runFileStreamOld = new RunFileStream(ctx, "left", status.branch[LEFT_PARTITION]);
        runFileStream = new RunFileStream(ctx, "left", status.branch[LEFT_PARTITION]);

        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.fine(
                    "MergeJoiner has started partition " + partition + " with " + memorySize + " frames of memory.");
        }
    }

    private boolean addToMemory(ITupleAccessor accessor) throws HyracksDataException {
        TuplePointer tp = new TuplePointer();
        if (bufferManager.insertTuple(accessor, accessor.getTupleId(), tp)) {
            memoryBuffer.add(tp);
            return true;
        }
        return false;
    }

    private void removeFromMemory(TuplePointer tp) throws HyracksDataException {
        memoryBuffer.remove(tp);
        bufferManager.deleteTuple(tp);
    }

    private void addToResult(IFrameTupleAccessor accessorLeft, int leftTupleIndex, IFrameTupleAccessor accessorRight,
            int rightTupleIndex, IFrameWriter writer) throws HyracksDataException {
        FrameUtils.appendConcatToWriter(writer, resultAppender, accessorLeft, leftTupleIndex, accessorRight,
                rightTupleIndex);
        joinResultCount++;
    }

    private void flushMemory() throws HyracksDataException {
        memoryBuffer.clear();
        bufferManager.reset();
    }

    // memory management
    private boolean memoryHasTuples() {
        return bufferManager.getNumTuples() > 0;
    }

    /**
     * Ensures a frame exists for the right branch, either from memory or the run file.
     *
     * @throws HyracksDataException
     */
    private TupleStatus loadRightTuple() throws HyracksDataException {
        TupleStatus loaded = loadMemoryTuple(RIGHT_PARTITION);
        if (loaded == TupleStatus.UNKNOWN) {
            loaded = pauseAndLoadRightTuple();
        }
        return loaded;
    }

    /**
     * Ensures a frame exists for the right branch, either from memory or the run file.
     *
     * @throws HyracksDataException
     */
    private TupleStatus loadLeftTuple() throws HyracksDataException {
        TupleStatus loaded;
        if (status.branch[LEFT_PARTITION].isRunFileReading()) {
            loaded = loadSpilledTuple(LEFT_PARTITION);
            if (loaded.isEmpty()) {
                if (status.branch[LEFT_PARTITION].isRunFileWriting() && !status.branch[LEFT_PARTITION].hasMore()) {
                    unfreezeAndContinue(inputAccessor[LEFT_PARTITION]);
                } else {
                    continueStream(inputAccessor[LEFT_PARTITION]);
                }
                loaded = loadLeftTuple();
            }
        } else {
            loaded = loadMemoryTuple(LEFT_PARTITION);
        }
        return loaded;
    }

    private TupleStatus loadSpilledTuple(int partition) throws HyracksDataException {
        if (!inputAccessor[partition].exists()) {
            runFileStream.loadNextBuffer(tmpAccessor);
            if (!runFileStreamOld.loadNextBuffer(inputAccessor[partition])) {
                return TupleStatus.EMPTY;
            }
        }
        return TupleStatus.LOADED;
    }

    /**
     * Left
     *
     * @throws HyracksDataException
     */
    @Override
    public void processLeftFrame(IFrameWriter writer) throws HyracksDataException {
        TupleStatus leftTs = loadLeftTuple();
        TupleStatus rightTs = loadRightTuple();
        while (leftTs.isLoaded() && (status.branch[RIGHT_PARTITION].hasMore() || memoryHasTuples())) {
            if (status.branch[LEFT_PARTITION].isRunFileWriting()) {
                // Left side from disk
                leftTs = processLeftTupleSpill(writer);
            } else if (rightTs.isLoaded()
                    && mjc.checkToLoadNextRightTuple(inputAccessor[LEFT_PARTITION], inputAccessor[RIGHT_PARTITION])) {
                // Right side from stream
                processRightTuple();
                rightTs = loadRightTuple();
            } else {
                // Left side from stream
                processLeftTuple(writer);
                leftTs = loadLeftTuple();
            }
        }
    }

    @Override
    public void processLeftClose(IFrameWriter writer) throws HyracksDataException {
        if (status.branch[LEFT_PARTITION].isRunFileWriting()) {
            unfreezeAndContinue(inputAccessor[LEFT_PARTITION]);
        }
        processLeftFrame(writer);
        resultAppender.write(writer, true);
        if (LOGGER.isLoggable(Level.WARNING)) {
            LOGGER.warning("MergeJoiner statitics: " + joinComparisonCount + " comparisons, " + joinResultCount
                    + " results, " + spillCount + " spills, " + runFileStreamOld.getFileCount() + " files, "
                    + runFileStreamOld.getWriteCount() + " spill frames written, " + runFileStreamOld.getReadCount()
                    + " spill frames read.");
        }
    }

    private TupleStatus processLeftTupleSpill(IFrameWriter writer) throws HyracksDataException {
        //        System.err.print("Spill ");

        runFileStreamOld.addToRunFile(inputAccessor[LEFT_PARTITION]);
        if (true) {
            runFileStream.addToRunFile(inputAccessor[LEFT_PARTITION]);
        }

        processLeftTuple(writer);

        // Memory is empty and we can start processing the run file.
        if (!memoryHasTuples() && status.branch[LEFT_PARTITION].isRunFileWriting()) {
            unfreezeAndContinue(inputAccessor[LEFT_PARTITION]);
        }
        return loadLeftTuple();
    }

    private void processLeftTuple(IFrameWriter writer) throws HyracksDataException {
        //        TuplePrinterUtil.printTuple("Left", inputAccessor[LEFT]);
        // Check against memory (right)
        if (memoryHasTuples()) {
            for (int i = memoryBuffer.size() - 1; i > -1; --i) {
                memoryAccessor.reset(memoryBuffer.get(i));
                if (mjc.checkToSaveInResult(inputAccessor[LEFT_PARTITION], inputAccessor[LEFT_PARTITION].getTupleId(),
                        memoryAccessor, memoryBuffer.get(i).getTupleIndex(), false)) {
                    // add to result
                    addToResult(inputAccessor[LEFT_PARTITION], inputAccessor[LEFT_PARTITION].getTupleId(),
                            memoryAccessor, memoryBuffer.get(i).getTupleIndex(), writer);
                }
                joinComparisonCount++;
                if (mjc.checkToRemoveInMemory(inputAccessor[LEFT_PARTITION], inputAccessor[LEFT_PARTITION].getTupleId(),
                        memoryAccessor, memoryBuffer.get(i).getTupleIndex())) {
                    // remove from memory
                    //                    TuplePrinterUtil.printTuple("Remove Memory", memoryAccessor, memoryBuffer.get(i).getTupleIndex());
                    removeFromMemory(memoryBuffer.get(i));
                }
            }
        }
        inputAccessor[LEFT_PARTITION].next();
    }

    private void processRightTuple() throws HyracksDataException {
        // append to memory
        if (mjc.checkToSaveInMemory(inputAccessor[LEFT_PARTITION], inputAccessor[RIGHT_PARTITION])) {
            if (!addToMemory(inputAccessor[RIGHT_PARTITION])) {
                // go to log saving state
                freezeAndSpill();
                return;
            }
        }
        //        TuplePrinterUtil.printTuple("Memory", inputAccessor[RIGHT]);
        inputAccessor[RIGHT_PARTITION].next();
    }

    private void freezeAndSpill() throws HyracksDataException {
        //        System.err.println("freezeAndSpill");
        if (LOGGER.isLoggable(Level.WARNING)) {
            LOGGER.warning("freeze snapshot: " + frameCounts[RIGHT_PARTITION] + " right, " + frameCounts[LEFT_PARTITION]
                    + " left, " + joinComparisonCount + " comparisons, " + joinResultCount + " results, ["
                    + bufferManager.getNumTuples() + " tuples memory].");
        }

        if (runFilePointer.getFileOffset() > 0) {

        } else {
            runFilePointer.reset(0, 0);
            runFileStream.startRunFileWriting();
        }
        runFileStreamOld.startRunFileWriting();
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.fine(
                    "Memory is full. Freezing the right branch. (memory tuples: " + bufferManager.getNumTuples() + ")");
        }
        spillCount++;
    }

    private void continueStream(ITupleAccessor accessor) throws HyracksDataException {
        //        System.err.println("continueStream");

        runFileStreamOld.closeRunFileReading();
        accessor.reset(inputBuffer[LEFT_PARTITION]);
        accessor.setTupleId(leftStreamIndex);
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.fine("Continue with left stream.");
        }
    }

    private void unfreezeAndContinue(ITupleAccessor accessor) throws HyracksDataException {
        //        System.err.println("unfreezeAndContinue");
        if (LOGGER.isLoggable(Level.WARNING)) {
            LOGGER.warning("snapshot: " + frameCounts[RIGHT_PARTITION] + " right, " + frameCounts[LEFT_PARTITION]
                    + " left, " + joinComparisonCount + " comparisons, " + joinResultCount + " results, ["
                    + bufferManager.getNumTuples() + " tuples memory, " + spillCount + " spills, "
                    + (runFileStreamOld.getFileCount() - spillFileCount) + " files, "
                    + (runFileStreamOld.getWriteCount() - spillWriteCount) + " written, "
                    + (runFileStreamOld.getReadCount() - spillReadCount) + " read].");
            spillFileCount = runFileStreamOld.getFileCount();
            spillReadCount = runFileStreamOld.getReadCount();
            spillWriteCount = runFileStreamOld.getWriteCount();
        }

        runFileStreamOld.flushAndStopRunFile(accessor);
        runFileStream.flushAndStopRunFile(accessor);
        flushMemory();
        if (!status.branch[LEFT_PARTITION].isRunFileReading()) {
            leftStreamIndex = accessor.getTupleId();
        }
        runFileStreamOld.startReadingRunFile(accessor);

        runFileStream.resetReadPointer(runFilePointer.getFileOffset());
        accessor.setTupleId(runFilePointer.getTupleIndex());
        runFileStream.startReadingRunFile(accessor);
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.fine("Unfreezing right partition.");
        }

    }

}