/**
 * 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.tajo.engine.planner.physical;

import com.google.common.base.Preconditions;
import com.google.common.primitives.*;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.IOUtils;
import org.apache.tajo.BuiltinStorages;
import org.apache.tajo.SessionVars;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.datum.TextDatum;
import org.apache.tajo.engine.planner.PhysicalPlanningException;
import org.apache.tajo.engine.query.QueryContext;
import org.apache.tajo.exception.TajoRuntimeException;
import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.plan.logical.ScanNode;
import org.apache.tajo.plan.logical.SortNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.Scanner;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.storage.fragment.FragmentConvertor;
import org.apache.tajo.storage.rawfile.DirectRawFileWriter;
import org.apache.tajo.tuple.memory.OffHeapRowBlockUtils;
import org.apache.tajo.tuple.memory.UnSafeTuple;
import org.apache.tajo.tuple.memory.UnSafeTupleList;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.FileUtil;
import org.apache.tajo.worker.TaskAttemptContext;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * This external sort algorithm can be characterized by the followings:
 *
 * <ul>
 *   <li>in-memory sort if input data size fits a sort buffer</li>
 *   <li>k-way merge sort if input data size exceeds the size of sort buffer</li>
 *   <li>parallel merge</li>
 *   <li>final merge avoidance</li>
 *   <li>Unbalance merge if needed</li>
 * </ul>
 */
public class ExternalSortExec extends SortExec {

  enum SortAlgorithm{
    TIM,
    MSD_RADIX,
  }

  /** Class logger */
  private static final Log LOG = LogFactory.getLog(ExternalSortExec.class);
  /** The prefix of fragment name for intermediate */
  private static final String INTERMEDIATE_FILE_PREFIX = "@interFile_";

  private SortNode plan;
  /** the data format of intermediate file*/
  private TableMeta intermediateMeta;
  /** the defaultFanout of external sort */
  private final int defaultFanout;
  /** It's the size of in-memory table. If memory consumption exceeds it, store the memory table into a disk. */
  private final long sortBufferBytesNum;
  /** the number of available cores */
  private final int allocatedCoreNum;
  /** If there are available multiple cores, it tries parallel merge. */
  private ExecutorService executorService;
  /** used for in-memory sort of each chunk. */
  private UnSafeTupleList inMemoryTable;
  /** for zero copy tuple comparison */
  private Comparator<UnSafeTuple> unSafeComparator;
  /** for other type tuple comparison */
  private Comparator<Tuple> primitiveComparator;
  /** temporal dir */
  private Path sortTmpDir;
  /** It enables round-robin disks allocation */
  private final LocalDirAllocator localDirAllocator;
  /** local file system */
  private final RawLocalFileSystem localFS;
  /** final output files which are used for cleaning */
  private List<Chunk> finalOutputFiles = null;
  /** for directly merging sorted inputs */
  private List<Chunk> mergedInputFragments = null;

  ///////////////////////////////////////////////////
  // transient variables
  ///////////////////////////////////////////////////
  /** already sorted or not */
  private boolean sorted = false;
  /** the final result */
  private Scanner result;
  /** total bytes of input data */
  private long inputBytes;

  private final SortAlgorithm sortAlgorithm;

  private ExternalSortExec(final TaskAttemptContext context, final SortNode plan)
      throws PhysicalPlanningException {
    super(context, plan.getInSchema(), plan.getOutSchema(), null, plan.getSortKeys());

    this.plan = plan;
    this.defaultFanout = context.getConf().getIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT);
    if (defaultFanout < 2) {
      throw new PhysicalPlanningException(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT.varname + " cannot be lower than 2");
    }
    // TODO - sort buffer and core num should be changed to use the allocated container resource.
    this.sortBufferBytesNum = context.getQueryContext().getInt(SessionVars.EXTSORT_BUFFER_SIZE) * StorageUnit.MB;
    this.allocatedCoreNum = context.getConf().getIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_THREAD_NUM);
    this.localDirAllocator = new LocalDirAllocator(ConfVars.WORKER_TEMPORAL_DIR.varname);
    this.localFS = new RawLocalFileSystem();
    this.intermediateMeta = CatalogUtil.newTableMeta(BuiltinStorages.DRAW, context.getConf());
    this.inputStats = new TableStats();
    this.sortAlgorithm = getSortAlgorithm(context.getQueryContext(), sortSpecs);
    LOG.info(sortAlgorithm.name() + " sort is selected");
  }

  private static SortAlgorithm getSortAlgorithm(QueryContext context, SortSpec[] sortSpecs) {
    String sortAlgorithm = context.get(SessionVars.SORT_ALGORITHM, SortAlgorithm.TIM.name());
    if (Arrays.stream(sortSpecs)
        .filter(sortSpec -> !RadixSort.isApplicableType(sortSpec)).count() > 0) {
      if (sortAlgorithm.equalsIgnoreCase(SortAlgorithm.MSD_RADIX.name())) {
        LOG.warn("Non-applicable types exist. Falling back to " + SortAlgorithm.TIM.name() + " sort");
      }
      return SortAlgorithm.TIM;
    }
    if (sortAlgorithm.equalsIgnoreCase(SortAlgorithm.TIM.name())) {
      return SortAlgorithm.TIM;
    } else if (sortAlgorithm.equalsIgnoreCase(SortAlgorithm.MSD_RADIX.name())) {
      return SortAlgorithm.MSD_RADIX;
    } else {
      LOG.warn("Unknown sort type: " + sortAlgorithm);
      LOG.warn("Falling back to " + SortAlgorithm.TIM.name() + " sort");
      return SortAlgorithm.TIM;
    }
  }

  public ExternalSortExec(final TaskAttemptContext context,final SortNode plan, final ScanNode scanNode,
                          final CatalogProtos.FragmentProto[] fragments) throws PhysicalPlanningException {
    this(context, plan);

    mergedInputFragments = new ArrayList<>();
    for (CatalogProtos.FragmentProto proto : fragments) {
      FileFragment fragment = FragmentConvertor.convert(FileFragment.class, proto);
      mergedInputFragments.add(new Chunk(inSchema, fragment, scanNode.getTableDesc().getMeta()));
    }
  }

  public ExternalSortExec(final TaskAttemptContext context, final SortNode plan, final PhysicalExec child)
      throws IOException {
    this(context, plan);
    setChild(child);
  }

  @Override
  public void init() throws IOException {
    if(allocatedCoreNum > 1) {
      this.executorService = Executors.newFixedThreadPool(this.allocatedCoreNum);
    }

    this.sortTmpDir = getExecutorTmpDir();

    int initialArraySize = context.getQueryContext().getInt(SessionVars.SORT_LIST_SIZE);
    this.inMemoryTable = new UnSafeTupleList(SchemaUtil.toDataTypes(inSchema), initialArraySize);
    this.unSafeComparator = new UnSafeComparator(inSchema, sortSpecs);
    this.primitiveComparator = new PrimitiveComparator(inSchema, sortSpecs);

    super.init();
  }

  public SortNode getPlan() {
    return this.plan;
  }

  private List<UnSafeTuple> sort(UnSafeTupleList tupleBlock) {
    switch (sortAlgorithm) {
      case TIM:
        return OffHeapRowBlockUtils.sort(tupleBlock, unSafeComparator);
      case MSD_RADIX:
        return RadixSort.sort(context.getQueryContext(), tupleBlock, inSchema, sortSpecs, unSafeComparator);
      default:
        // The below line is not reachable. So, an exception should be thrown if it is executed.
        throw new TajoRuntimeException(new UnsupportedException(sortAlgorithm.name()));
    }
  }

  /**
   * Sort a tuple block and store them into a chunk file
   */
  private Chunk sortAndStoreChunk(int chunkId, UnSafeTupleList tupleBlock)
      throws IOException {
    int rowNum = tupleBlock.size();

    long sortStart = System.currentTimeMillis();
    this.sort(tupleBlock);
    long sortEnd = System.currentTimeMillis();

    long chunkWriteStart = System.currentTimeMillis();
    Path outputPath = getChunkPathForWrite(0, chunkId);
    final DirectRawFileWriter appender =
        new DirectRawFileWriter(context.getConf(), null, inSchema, intermediateMeta, outputPath);
    appender.init();
    for (Tuple t : tupleBlock) {
      appender.addTuple(t);
    }
    appender.close();
    long chunkWriteEnd = System.currentTimeMillis();

    info(LOG, "Chunk #" + chunkId + " sort and written (" +
        FileUtil.humanReadableByteCount(appender.getOffset(), false) + " bytes, " + rowNum + " rows, " +
        "sort time: " + (sortEnd - sortStart) + " msec, " +
        "write time: " + (chunkWriteEnd - chunkWriteStart) + " msec)");

    FileFragment frag = new FileFragment("", outputPath, 0,
        new File(localFS.makeQualified(outputPath).toUri()).length());
    return new Chunk(inSchema, frag, intermediateMeta);
  }

  /**
   * It divides all tuples into a number of chunks, then sort for each chunk.
   *
   * @return All paths of chunks
   * @throws java.io.IOException
   */
  private List<Chunk> sortAndStoreAllChunks() throws IOException {
    Tuple tuple;
    List<Chunk> chunkPaths = new ArrayList<>();

    int chunkId = 0;
    long runStartTime = System.currentTimeMillis();

    while (!context.isStopped() && (tuple = child.next()) != null) { // partition sort start
      inMemoryTable.addTuple(tuple);

      if (inMemoryTable.usedMem() > sortBufferBytesNum) { // if input data exceeds main-memory at least once
        long runEndTime = System.currentTimeMillis();
        info(LOG, "Chunk #" + chunkId + " run loading time: " + (runEndTime - runStartTime) + " msec");
        runStartTime = runEndTime;

        info(LOG, "Memory consumption exceeds " + FileUtil.humanReadableByteCount(inMemoryTable.usedMem(), false));

        chunkPaths.add(sortAndStoreChunk(chunkId, inMemoryTable));
        inMemoryTable.clear();
        chunkId++;

        // When the volume of sorting data once exceed the size of sort buffer,
        // the total progress of this external sort is divided into two parts.
        // In contrast, if the data fits in memory, the progress is only one part.
        //
        // When the progress is divided into two parts, the first part sorts tuples on memory and stores them
        // into a chunk. The second part merges stored chunks into fewer chunks, and it continues until the number
        // of merged chunks is fewer than the default fanout.
        //
        // The fact that the code reach here means that the first chunk has been just stored.
        // That is, the progress was divided into two parts.
        // So, it multiply the progress of the children operator and 0.5f.
        progress = child.getProgress() * 0.5f;
      }
    }

    if(inMemoryTable.size() > 0) { //if there are at least one or more input tuples
      //store the remain data into a memory chunk.
      chunkPaths.add(new Chunk(inSchema, inMemoryTable, intermediateMeta));
    }

    // get total loaded (or stored) bytes and total row numbers
    TableStats childTableStats = child.getInputStats();
    if (childTableStats != null) {
      inputBytes = childTableStats.getNumBytes();
    }
    return chunkPaths;
  }

  /**
   * Get a local path from all temporal paths in round-robin manner.
   */
  private synchronized Path getChunkPathForWrite(int level, int chunkId) throws IOException {
    return localFS.makeQualified(localDirAllocator.getLocalPathForWrite(
        sortTmpDir + "/" + level + "_" + chunkId, context.getConf()));
  }

  @Override
  public Tuple next() throws IOException {

    if (!sorted) { // if not sorted, first sort all data

      // if input files are given, it starts merging directly.
      if (mergedInputFragments != null) {
        try {
          this.result = externalMergeAndSort(mergedInputFragments);
          this.inputBytes = result.getInputStats().getNumBytes();
        } catch (Exception e) {
          throw new PhysicalPlanningException(e);
        }
      } else {
        // Try to sort all data, and store them as multiple chunks if memory exceeds
        long startTimeOfChunkSplit = System.currentTimeMillis();
        List<Chunk> chunks = sortAndStoreAllChunks();
        long endTimeOfChunkSplit = System.currentTimeMillis();
        info(LOG, chunks.size() + " Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec");

        if(chunks.size() == 0) {
          this.result = new NullScanner(context.getConf(), inSchema, intermediateMeta, null);
        } else {
          try {
            this.result = externalMergeAndSort(chunks);
          } catch (Exception e) {
            throw new PhysicalPlanningException(e);
          }
        }
      }

      sorted = true;
      result.init();

      // if loaded and sorted, we assume that it proceeds the half of one entire external sort operation.
      progress = 0.5f;
    }

    return result.next();
  }

  private int calculateFanout(int remainInputChunks, int inputNum, int outputNum, int startIdx) {
    int computedFanout = Math.min(remainInputChunks, defaultFanout);

    // Why should we detect an opportunity for unbalanced merge?
    //
    // Assume that a fanout is given by 8 and there are 10 chunks.
    // If we firstly merge 3 chunks into one chunk, there remain only 8 chunks.
    // Then, we can just finish the merge phase even though we don't complete merge phase on all chunks.
    if (checkIfCanBeUnbalancedMerged(inputNum - (startIdx + computedFanout), outputNum + 1)) {
      int candidateFanout = computedFanout;
      while (checkIfCanBeUnbalancedMerged(inputNum - (startIdx + candidateFanout), outputNum + 1)) {
        candidateFanout--;
      }
      int beforeFanout = computedFanout;
      if (computedFanout > candidateFanout + 1) {
        computedFanout = candidateFanout + 1;
        info(LOG, "Fanout reduced for unbalanced merge: " + beforeFanout + " -> " + computedFanout);
      }
    }

    return computedFanout;
  }

  private Scanner externalMergeAndSort(List<Chunk> chunks) throws Exception {
    int level = 0;
    final List<Chunk> inputFiles = new ArrayList<>(chunks);
    final List<Chunk> outputFiles = new ArrayList<>();
    int remainRun = inputFiles.size();
    int chunksSize = chunks.size();

    long mergeStart = System.currentTimeMillis();

    // continue until the remain runs are larger than defaultFanout
    while (remainRun > defaultFanout) {

      // reset outChunkId
      int remainInputRuns = inputFiles.size();
      int outChunkId = 0;
      int outputFileNum = 0;
      List<Future<Chunk>> futures = new ArrayList<>();
      // the number of files being merged in threads.
      List<Integer> numberOfMergingFiles = new ArrayList<>();

      for (int startIdx = 0; startIdx < inputFiles.size();) {

        // calculate proper fanout
        int fanout = calculateFanout(remainInputRuns, inputFiles.size(), outputFileNum, startIdx);
        // how many files are merged in ith thread?
        numberOfMergingFiles.add(fanout);
        // launch a merger runner
        if(allocatedCoreNum > 1) {
          futures.add(executorService.submit(
              new KWayMergerCaller(level, outChunkId++, inputFiles, startIdx, fanout, false)));
        } else {
          final SettableFuture<Chunk> future = SettableFuture.create();
          future.set(new KWayMergerCaller(level, outChunkId++, inputFiles, startIdx, fanout, false).call());
          futures.add(future);
        }
        outputFileNum++;

        startIdx += fanout;
        remainInputRuns = inputFiles.size() - startIdx;

        // If unbalanced merge is available, it finishes the merge phase earlier.
        if (checkIfCanBeUnbalancedMerged(remainInputRuns, outputFileNum)) {
          info(LOG, "Unbalanced merge possibility detected: number of remain input (" + remainInputRuns
              + ") and output files (" + outputFileNum + ") <= " + defaultFanout);

          List<Chunk> switched = new ArrayList<>();
          // switch the remain inputs to the next outputs
          for (int j = startIdx; j < inputFiles.size(); j++) {
            switched.add(inputFiles.get(j));
          }
          inputFiles.removeAll(switched);
          outputFiles.addAll(switched);

          break;
        }
      }

      // wait for all sort runners
      int finishedMerger = 0;
      int index = 0;
      for (Future<Chunk> future : futures) {
        outputFiles.add(future.get());
        // Getting the number of merged files
        finishedMerger += numberOfMergingFiles.get(index++);
        // progress = (# number of merged files / total number of files) * 0.5;
        progress = ((float)finishedMerger/(float)chunksSize) * 0.5f;
      }

      /*
       * delete merged intermediate files
       * 
       * There may be 4 different types of file fragments in the list inputFiles
       * + A: a fragment created from fetched data from a remote host. By default, this fragment represents
       * a whole physical file (i.e., startOffset == 0 and length == length of physical file)
       * + B1: a fragment created from a local file (pseudo-fetched data from local host) in which the fragment
       * represents the whole physical file (i.e., startOffset == 0 AND length == length of physical file)
       * + B2: a fragment created from a local file (pseudo-fetched data from local host) in which the fragment
       * represents only a part of the physical file (i.e., startOffset > 0 OR length != length of physical file)
       * + C: a fragment created from merging some fragments of the above types. When this fragment is created,
       * its startOffset is set to 0 and its length is set to the length of the physical file, automatically
       * 
       * Fragments of types A, B1, and B2 are inputs of ExternalSortExec. Among them, only B2-type fragments will
       * possibly be used by another task in the future. Thus, ideally, all fragments of types A, B1, and C can be
       * deleted at this point. However, for the ease of future code maintenance, we delete only type-C fragments here
       */
      int numDeletedFiles = 0;
      for (Chunk chunk : inputFiles) {
        if (chunk.isMemory()) {
          if (LOG.isDebugEnabled()) {
            debug(LOG, "Remove intermediate memory tuples: " + chunk.getMemoryTuples().usedMem());
          }
          chunk.getMemoryTuples().release();
        } else if (chunk.getFragment().getTableName().contains(INTERMEDIATE_FILE_PREFIX)) {
          localFS.delete(chunk.getFragment().getPath(), true);
          numDeletedFiles++;

          if (LOG.isDebugEnabled()) {
            debug(LOG, "Delete merged intermediate file: " + chunk.getFragment());
          }
        }
      }
      if(LOG.isDebugEnabled()) {
        debug(LOG, numDeletedFiles + " merged intermediate files deleted");
      }

      // switch input files to output files, and then clear outputFiles
      inputFiles.clear();
      inputFiles.addAll(outputFiles);
      remainRun = inputFiles.size();
      outputFiles.clear();
      level++;
    }

    long mergeEnd = System.currentTimeMillis();
    info(LOG, "Total merge time: " + (mergeEnd - mergeStart) + " msec");

    // final result
    finalOutputFiles = inputFiles;

    result = createFinalMerger(inputFiles);
    return result;
  }

  /**
   * Merge Thread
   */
  private class KWayMergerCaller implements Callable<Chunk> {
    final int level;
    final int nextRunId;
    final List<Chunk> inputFiles;
    final int startIdx;
    final int mergeFanout;
    final boolean updateInputStats;

    public KWayMergerCaller(final int level, final int nextRunId, final List<Chunk> inputFiles,
                            final int startIdx, final int mergeFanout, final boolean updateInputStats) {
      this.level = level;
      this.nextRunId = nextRunId;
      this.inputFiles = inputFiles;
      this.startIdx = startIdx;
      this.mergeFanout = mergeFanout;
      this.updateInputStats = updateInputStats;
    }

    @Override
    public Chunk call() throws Exception {
      final Path outputPath = getChunkPathForWrite(level + 1, nextRunId);
      info(LOG, mergeFanout + " files are being merged to an output file " + outputPath.getName());
      long mergeStartTime = System.currentTimeMillis();

      final Scanner merger = createKWayMerger(inputFiles, startIdx, mergeFanout);
      merger.init();

      final DirectRawFileWriter output =
          new DirectRawFileWriter(context.getConf(), null, inSchema, intermediateMeta, outputPath);
      output.init();

      Tuple mergeTuple;
      while((mergeTuple = merger.next()) != null) {
        output.addTuple(mergeTuple);
      }
      merger.close();
      output.close();
      long mergeEndTime = System.currentTimeMillis();
      info(LOG, outputPath.getName() + " is written to a disk. ("
          + FileUtil.humanReadableByteCount(output.getOffset(), false)
          + " bytes, " + (mergeEndTime - mergeStartTime) + " msec)");
      File f = new File(localFS.makeQualified(outputPath).toUri());
      FileFragment frag = new FileFragment(INTERMEDIATE_FILE_PREFIX + outputPath.getName(), outputPath, 0, f.length());
      return new Chunk(inSchema, frag, intermediateMeta);
    }
  }

  /**
   * It checks if unbalanced merge is possible.
   */
  private boolean checkIfCanBeUnbalancedMerged(int remainInputNum, int outputNum) {
    return (remainInputNum + outputNum) <= defaultFanout;
  }

  /**
   * Create a merged file scanner or k-way merge scanner.
   */
  private Scanner createFinalMerger(List<Chunk> inputs) throws IOException {
    if (inputs.size() == 1) {
      this.result = getScanner(inputs.get(0));
    } else {
      this.result = createKWayMerger(inputs, 0, inputs.size());
    }
    return result;
  }

  private Scanner getScanner(Chunk chunk) throws IOException {
    if (chunk.isMemory()) {
      long sortStart = System.currentTimeMillis();

      this.sort(inMemoryTable);
      Scanner scanner = new MemTableScanner<>(inMemoryTable, inMemoryTable.size(), inMemoryTable.usedMem());
      if(LOG.isDebugEnabled()) {
        debug(LOG, "Memory Chunk sort (" + FileUtil.humanReadableByteCount(inMemoryTable.usedMem(), false)
            + " bytes, " + inMemoryTable.size() + " rows, sort time: "
            + (System.currentTimeMillis() - sortStart) + " msec)");
      }
      return scanner;
    } else {
      return TablespaceManager.getLocalFs().getScanner(chunk.meta, chunk.schema, chunk.fragment, chunk.schema);
    }
  }

  private Scanner createKWayMerger(List<Chunk> inputs, final int startChunkId, final int num) throws IOException {
    final Scanner [] sources = new Scanner[num];
    for (int i = 0; i < num; i++) {
      sources[i] = getScanner(inputs.get(startChunkId + i));
    }
    return createKWayMergerInternal(sources, 0, num);
  }

  private Scanner createKWayMergerInternal(final Scanner [] sources, final int startIdx, final int num)
      throws IOException {
    if (num > 1) {
      final int mid = (int) Math.ceil((float)num / 2);
      Scanner left = createKWayMergerInternal(sources, startIdx, mid);
      Scanner right = createKWayMergerInternal(sources, startIdx + mid, num - mid);
      return new PairWiseMerger(inSchema, left, right, primitiveComparator);
    } else {
      return sources[startIdx];
    }
  }

  private static class MemTableScanner<T extends Tuple> extends AbstractScanner {
    final Iterable<T> iterable;
    final long sortAndStoredBytes;
    final int totalRecords;

    Iterator<T> iterator;
    // for input stats
    float scannerProgress;
    int numRecords;
    TableStats scannerTableStats;

    public MemTableScanner(Iterable<T> iterable, int length, long inBytes) {
      this.iterable = iterable;
      this.totalRecords = length;
      this.sortAndStoredBytes = inBytes;
    }

    @Override
    public void init() throws IOException {
      iterator = iterable.iterator();

      scannerProgress = 0.0f;
      numRecords = 0;

      // it will be returned as the final stats
      scannerTableStats = new TableStats();
      scannerTableStats.setNumBytes(sortAndStoredBytes);
      scannerTableStats.setReadBytes(sortAndStoredBytes);
      scannerTableStats.setNumRows(totalRecords);
    }

    @Override
    public Tuple next() throws IOException {
      if (iterator.hasNext()) {
        numRecords++;
        return iterator.next();
      } else {
        return null;
      }
    }

    @Override
    public void reset() throws IOException {
      init();
    }

    @Override
    public void close() throws IOException {
      iterator = null;
      scannerProgress = 1.0f;
    }

    @Override
    public float getProgress() {
      if (iterator != null && numRecords > 0) {
        return (float)numRecords / (float)totalRecords;

      } else { // if an input is empty
        return scannerProgress;
      }
    }

    @Override
    public TableStats getInputStats() {
      return scannerTableStats;
    }
  }

  enum State {
    NEW,
    INITED,
    CLOSED
  }

  /**
   * Two-way merger scanner that reads two input sources and outputs one output tuples sorted in some order.
   */
  private static class PairWiseMerger extends AbstractScanner {

    protected final Schema schema;
    protected final Comparator<Tuple> comparator;

    protected final Scanner leftScan;
    protected final Scanner rightScan;

    private Tuple leftTuple;
    private Tuple rightTuple;
    private boolean leftEOF;
    private boolean rightEOF;

    private Tuple outTuple;

    private float mergerProgress;
    private TableStats mergerInputStats;

    private State state = State.NEW;

    public PairWiseMerger(Schema schema, Scanner leftScanner, Scanner rightScanner, Comparator<Tuple> comparator)
        throws IOException {
      this.schema = schema;
      this.leftScan = leftScanner;
      this.rightScan = rightScanner;
      this.comparator = comparator;
    }

    private void setState(State state) {
      this.state = state;
    }

    @Override
    public void init() throws IOException {
      if (state == State.NEW) {
        leftScan.init();
        rightScan.init();

        mergerInputStats = new TableStats();
        mergerProgress = 0.0f;

        setState(State.INITED);
      } else {
        throw new IllegalStateException("Illegal State: init() is not allowed in " + state.name());
      }
    }

    protected int compare() {
      return comparator.compare(leftTuple, rightTuple);
    }

    @Override
    public Tuple next() throws IOException {
      if(!leftEOF && leftTuple == null) {
        leftTuple = leftScan.next();
      }

      if(!rightEOF && rightTuple == null) {
        rightTuple = rightScan.next();
      }

      if (leftTuple != null && rightTuple != null) {
        if (compare() < 0) {
          outTuple = leftTuple;
          leftTuple = null;
        } else {
          outTuple = rightTuple;
          rightTuple = null;
        }
        return outTuple;
      }

      if (leftTuple == null) {
        leftEOF = true;

        if (rightTuple != null) {
          outTuple = rightTuple;
          rightTuple = null;
        } else {
          rightEOF = true;
          outTuple = null;
        }
      } else {
        rightEOF = true;
        outTuple = leftTuple;
        leftTuple = null;
      }
      return outTuple;
    }

    @Override
    public void reset() throws IOException {
      if (state == State.INITED) {
        leftScan.reset();
        rightScan.reset();

        leftTuple = null;
        rightTuple = null;
        outTuple = null;

        leftEOF = false;
        rightEOF = false;

      } else {
        throw new IllegalStateException("Illegal State: init() is not allowed in " + state.name());
      }
    }

    @Override
    public void close() throws IOException {
      IOUtils.cleanup(LOG, leftScan, rightScan);
      getInputStats();
      mergerProgress = 1.0f;
      leftTuple = null;
      rightTuple = null;
      setState(State.CLOSED);
    }

    @Override
    public Schema getSchema() {
      return schema;
    }

    @Override
    public float getProgress() {
      if (leftScan == null) {
        return mergerProgress;
      }
      return leftScan.getProgress() * 0.5f + rightScan.getProgress() * 0.5f;
    }

    @Override
    public TableStats getInputStats() {
      if (leftScan == null) {
        return mergerInputStats;
      }
      TableStats leftInputStats = leftScan.getInputStats();
      if (mergerInputStats == null) {
        mergerInputStats = new TableStats();
      }
      mergerInputStats.setNumBytes(0);
      mergerInputStats.setReadBytes(0);
      mergerInputStats.setNumRows(0);

      if (leftInputStats != null) {
        mergerInputStats.setNumBytes(leftInputStats.getNumBytes());
        mergerInputStats.setReadBytes(leftInputStats.getReadBytes());
        mergerInputStats.setNumRows(leftInputStats.getNumRows());
      }

      TableStats rightInputStats = rightScan.getInputStats();
      if (rightInputStats != null) {
        mergerInputStats.setNumBytes(mergerInputStats.getNumBytes() + rightInputStats.getNumBytes());
        mergerInputStats.setReadBytes(mergerInputStats.getReadBytes() + rightInputStats.getReadBytes());
        mergerInputStats.setNumRows(mergerInputStats.getNumRows() + rightInputStats.getNumRows());
      }

      return mergerInputStats;
    }
  }

  @Override
  public void close() throws IOException {
    super.close();

    if (result != null) {
      result.close();
    }

    if (finalOutputFiles != null) {
      for (Chunk chunk : finalOutputFiles) {
        if (!chunk.isMemory()) {
          FileFragment frag = chunk.getFragment();
          File tmpFile = new File(localFS.makeQualified(frag.getPath()).toUri());
          if (frag.getStartKey() == 0 && frag.getLength() == tmpFile.length()) {
            localFS.delete(frag.getPath(), true);
            if(LOG.isDebugEnabled()) {
              debug(LOG, "Delete file: " + frag);
            }
          }
        }
      }
    }

    if(inMemoryTable != null) {
      inMemoryTable.release();
      inMemoryTable = null;
    }

    if(executorService != null){
      executorService.shutdown();
      executorService = null;
    }

    plan = null;
  }

  @Override
  public void rescan() throws IOException {
    if (result != null) {
      result.reset();
    }
    super.rescan();
    progress = 0.5f;
  }

  @Override
  public float getProgress() {
    if (result != null) {
      return progress + result.getProgress() * 0.5f;
    } else {
      return progress;
    }
  }

  @Override
  public TableStats getInputStats() {
    if (result != null) {

      TableStats tableStats = result.getInputStats();
      inputStats.setNumRows(tableStats.getNumRows());
      inputStats.setNumBytes(inputBytes);
      inputStats.setReadBytes(tableStats.getReadBytes());
    }
    return inputStats;
  }

  private static class Chunk {
    private FileFragment fragment;
    private TableMeta meta;
    private Schema schema;
    private UnSafeTupleList memoryTuples;
    private boolean isMemory;

    public Chunk(Schema schema, FileFragment fragment, TableMeta meta) {
      this.schema = schema;
      this.fragment = fragment;
      this.meta = meta;
    }

    public Chunk(Schema schema, UnSafeTupleList tuples, TableMeta meta) {
      this.memoryTuples = tuples;
      this.isMemory = true;
      this.schema = schema;
      this.meta = meta;
    }

    public FileFragment getFragment() {
      return fragment;
    }

    public TableMeta getMeta() {
      return meta;
    }

    public UnSafeTupleList getMemoryTuples() {
      return memoryTuples;
    }

    public boolean isMemory() {
      return isMemory;
    }

    public Schema getSchema() {
      return schema;
    }
  }

  /**
   * The Comparator class for UnSafeTuples
   *
   * @see UnSafeTuple
   */
  static class UnSafeComparator implements Comparator<UnSafeTuple> {
    private final int[] sortKeyIds;
    private final TajoDataTypes.Type[] sortKeyTypes;
    private final boolean[] asc;
    private final boolean[] nullFirsts;

    /**
     * @param schema   The schema of input tuples
     * @param sortKeys The description of sort keys
     */
    public UnSafeComparator(Schema schema, SortSpec[] sortKeys) {
      Preconditions.checkArgument(sortKeys.length > 0,
          "At least one sort key must be specified.");

      this.sortKeyIds = new int[sortKeys.length];
      this.sortKeyTypes = new TajoDataTypes.Type[sortKeys.length];
      this.asc = new boolean[sortKeys.length];
      this.nullFirsts = new boolean[sortKeys.length];
      for (int i = 0; i < sortKeys.length; i++) {
        if (sortKeys[i].getSortKey().hasQualifier()) {
          this.sortKeyIds[i] = schema.getColumnId(sortKeys[i].getSortKey().getQualifiedName());
        } else {
          this.sortKeyIds[i] = schema.getColumnIdByName(sortKeys[i].getSortKey().getSimpleName());
        }

        this.asc[i] = sortKeys[i].isAscending();
        this.nullFirsts[i] = sortKeys[i].isNullsFirst();
        this.sortKeyTypes[i] = sortKeys[i].getSortKey().getDataType().getType();
      }
    }

    @Override
    public int compare(UnSafeTuple tuple1, UnSafeTuple tuple2) {
      for (int i = 0; i < sortKeyIds.length; i++) {
        int compare = OffHeapRowBlockUtils.compareColumn(tuple1, tuple2,
            sortKeyIds[i], sortKeyTypes[i], asc[i], nullFirsts[i]);

        if (compare != 0) {
          return compare;
        }
      }
      return 0;
    }
  }

  /**
   * The Comparator class for raw file
   */
  static class PrimitiveComparator implements Comparator<Tuple> {
    private final int[] sortKeyIds;
    private final TajoDataTypes.Type[] sortKeyTypes;
    private final boolean[] asc;
    private final boolean[] nullFirsts;

    /**
     * @param schema   The schema of input tuples
     * @param sortKeys The description of sort keys
     */
    public PrimitiveComparator(Schema schema, SortSpec[] sortKeys) {
      Preconditions.checkArgument(sortKeys.length > 0,
          "At least one sort key must be specified.");

      this.sortKeyIds = new int[sortKeys.length];
      this.sortKeyTypes = new TajoDataTypes.Type[sortKeys.length];
      this.asc = new boolean[sortKeys.length];
      this.nullFirsts = new boolean[sortKeys.length];
      for (int i = 0; i < sortKeys.length; i++) {
        if (sortKeys[i].getSortKey().hasQualifier()) {
          this.sortKeyIds[i] = schema.getColumnId(sortKeys[i].getSortKey().getQualifiedName());
        } else {
          this.sortKeyIds[i] = schema.getColumnIdByName(sortKeys[i].getSortKey().getSimpleName());
        }

        this.asc[i] = sortKeys[i].isAscending();
        this.nullFirsts[i] = sortKeys[i].isNullsFirst();
        this.sortKeyTypes[i] = sortKeys[i].getSortKey().getDataType().getType();
      }
    }

    @Override
    public int compare(Tuple tuple1, Tuple tuple2) {
      for (int i = 0; i < sortKeyIds.length; i++) {
        int compare = compareColumn(tuple1, tuple2,
            sortKeyIds[i], sortKeyTypes[i], asc[i], nullFirsts[i]);

        if (compare != 0) {
          return compare;
        }
      }
      return 0;
    }

    public int compareColumn(Tuple tuple1, Tuple tuple2, int index, TajoDataTypes.Type type,
                             boolean ascending, boolean nullFirst) {
      final boolean n1 = tuple1.isBlankOrNull(index);
      final boolean n2 = tuple2.isBlankOrNull(index);
      if (n1 && n2) {
        return 0;
      }

      if (n1 ^ n2) {
        return nullFirst ? (n1 ? -1 : 1) : (n1 ? 1 : -1);
      }

      int compare;
      switch (type) {
      case BOOLEAN:
        compare = Booleans.compare(tuple1.getBool(index), tuple2.getBool(index));
        break;
      case BIT:
        compare = tuple1.getByte(index) - tuple2.getByte(index);
        break;
      case INT1:
      case INT2:
        compare = Shorts.compare(tuple1.getInt2(index), tuple2.getInt2(index));
        break;
      case DATE:
      case INT4:
        compare = Ints.compare(tuple1.getInt4(index), tuple2.getInt4(index));
        break;
      case INET4:
        compare = UnsignedInts.compare(tuple1.getInt4(index), tuple2.getInt4(index));
        break;
      case TIME:
      case TIMESTAMP:
      case INT8:
        compare = Longs.compare(tuple1.getInt8(index), tuple2.getInt8(index));
        break;
      case FLOAT4:
        compare = Floats.compare(tuple1.getFloat4(index), tuple2.getFloat4(index));
        break;
      case FLOAT8:
        compare = Doubles.compare(tuple1.getFloat8(index), tuple2.getFloat8(index));
        break;
      case CHAR:
      case TEXT:
      case BLOB:
        compare = TextDatum.COMPARATOR.compare(tuple1.getBytes(index), tuple2.getBytes(index));
        break;
      default:
        throw new TajoRuntimeException(
            new UnsupportedException("unknown data type '" + type.name() + "'"));
      }
      return ascending ? compare : -compare;
    }
  }
}