/*
 * 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.carbondata.processing.loading.sort.unsafe.merger;

import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
import org.apache.carbondata.processing.loading.row.IntermediateSortTempRow;
import org.apache.carbondata.processing.loading.sort.CarbonPriorityQueue;
import org.apache.carbondata.processing.loading.sort.SortStepRowHandler;
import org.apache.carbondata.processing.loading.sort.unsafe.UnsafeCarbonRowPage;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.SortTempChunkHolder;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.UnsafeFinalMergePageHolder;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.UnsafeInmemoryHolder;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.UnsafeSortTempFileChunkHolder;
import org.apache.carbondata.processing.sort.sortdata.SortParameters;
import org.apache.carbondata.processing.sort.sortdata.TableFieldStat;

import org.apache.log4j.Logger;

public class UnsafeSingleThreadFinalSortFilesMerger extends CarbonIterator<Object[]> {
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(UnsafeSingleThreadFinalSortFilesMerger.class.getName());

  /**
   * fileCounter
   */
  private int fileCounter;

  /**
   * recordHolderHeap
   */
  private CarbonPriorityQueue<SortTempChunkHolder> recordHolderHeapLocal;

  private SortParameters parameters;
  private SortStepRowHandler sortStepRowHandler;
  /**
   * tempFileLocation
   */
  private String[] tempFileLocation;

  private String tableName;

  private boolean isStopProcess;

  public UnsafeSingleThreadFinalSortFilesMerger(SortParameters parameters,
      String[] tempFileLocation) {
    this.parameters = parameters;
    this.sortStepRowHandler = new SortStepRowHandler(parameters);
    this.tempFileLocation = tempFileLocation;
    this.tableName = parameters.getTableName();
  }

  /**
   * This method will be used to merger the merged files
   *
   */
  public void startFinalMerge(UnsafeCarbonRowPage[] rowPages,
      List<UnsafeInMemoryIntermediateDataMerger> merges) throws CarbonDataWriterException {
    // remove the spilled pages
    for (Iterator<UnsafeInMemoryIntermediateDataMerger> iter = merges.iterator();
         iter.hasNext(); ) {
      UnsafeInMemoryIntermediateDataMerger merger = iter.next();
      if (merger.isSpillDisk()) {
        // it has already been closed once the spill is finished, so no need to close it here
        iter.remove();
      }
    }
    startSorting(rowPages, merges);
  }

  /**
   * Below method will be used to start storing process This method will get
   * all the temp files present in sort temp folder then it will create the
   * record holder heap and then it will read first record from each file and
   * initialize the heap
   *
   */
  private void startSorting(UnsafeCarbonRowPage[] rowPages,
      List<UnsafeInMemoryIntermediateDataMerger> merges) throws CarbonDataWriterException {
    try {
      List<File> filesToMergeSort = getFilesToMergeSort();
      this.fileCounter = rowPages.length + filesToMergeSort.size() + merges.size();
      if (fileCounter == 0) {
        LOGGER.info("No files to merge sort");
        return;
      }
      LOGGER.info(String.format("Starting final merge of %d pages, including row pages: %d"
          + ", sort temp files: %d, intermediate merges: %d",
          this.fileCounter, rowPages.length, filesToMergeSort.size(), merges.size()));

      // create record holder heap
      createRecordHolderQueue();
      TableFieldStat tableFieldStat = new TableFieldStat(parameters);
      // iterate over file list and create chunk holder and add to heap
      LOGGER.info("Started adding first record from each page");
      for (final UnsafeCarbonRowPage rowPage : rowPages) {

        SortTempChunkHolder sortTempFileChunkHolder = new UnsafeInmemoryHolder(rowPage);

        // initialize
        sortTempFileChunkHolder.readRow();

        recordHolderHeapLocal.add(sortTempFileChunkHolder);
      }

      for (final UnsafeInMemoryIntermediateDataMerger merger : merges) {

        SortTempChunkHolder sortTempFileChunkHolder =
            new UnsafeFinalMergePageHolder(merger, tableFieldStat);

        // initialize
        sortTempFileChunkHolder.readRow();

        recordHolderHeapLocal.add(sortTempFileChunkHolder);
      }

      for (final File file : filesToMergeSort) {

        SortTempChunkHolder sortTempFileChunkHolder =
            new UnsafeSortTempFileChunkHolder(file, parameters, true, tableFieldStat);

        // initialize
        sortTempFileChunkHolder.readRow();

        recordHolderHeapLocal.add(sortTempFileChunkHolder);
      }

      LOGGER.info("Heap Size: " + this.recordHolderHeapLocal.size());
    } catch (Exception e) {
      LOGGER.error(e.getMessage(), e);
      throw new CarbonDataWriterException(e);
    }
  }

  private List<File> getFilesToMergeSort() {
    // this can be partitionId, bucketId or rangeId, let's call it rangeId
    final int rangeId = parameters.getRangeId();

    FileFilter fileFilter = new FileFilter() {
      public boolean accept(File pathname) {
        return pathname.getName().startsWith(tableName + '_' + rangeId);
      }
    };

    // get all the merged files
    List<File> files = new ArrayList<File>(tempFileLocation.length);
    for (String tempLoc : tempFileLocation) {
      File[] subFiles = new File(tempLoc).listFiles(fileFilter);
      if (null != subFiles && subFiles.length > 0) {
        files.addAll(Arrays.asList(subFiles));
      }
    }

    return files;
  }

  /**
   * This method will be used to create the heap which will be used to hold
   * the chunk of data
   */
  private void createRecordHolderQueue() {
    // creating record holder heap
    this.recordHolderHeapLocal = new CarbonPriorityQueue<>(fileCounter);
  }

  /**
   * This method will be used to get the sorted row in 3-parted format.
   * The row will feed the following writer process step.
   *
   * @return sorted row
   */
  public Object[] next() {
    if (hasNext()) {
      return sortStepRowHandler.convertIntermediateSortTempRowTo3Parted(getSortedRecordFromFile());
    } else {
      throw new NoSuchElementException("No more elements to return");
    }
  }

  /**
   * This method will be used to get the sorted record from file
   *
   * @return sorted record sorted record
   */
  private IntermediateSortTempRow getSortedRecordFromFile() throws CarbonDataWriterException {
    IntermediateSortTempRow row = null;

    // poll the top object from heap
    // heap maintains binary tree which is based on heap condition that will
    // be based on comparator we are passing the heap
    // when will call poll it will always delete root of the tree and then
    // it does trickel down operation complexity is log(n)
    SortTempChunkHolder poll = this.recordHolderHeapLocal.peek();

    // get the row from chunk
    row = poll.getRow();

    // check if there no entry present
    if (!poll.hasNext()) {
      // if chunk is empty then close the stream
      poll.close();
      recordHolderHeapLocal.poll();

      // change the file counter
      --this.fileCounter;

      // return row
      return row;
    }

    // read new row
    try {
      poll.readRow();
    } catch (Exception e) {
      throw new CarbonDataWriterException(e);
    }

    // maintain heap
    this.recordHolderHeapLocal.siftTopDown();

    // return row
    return row;
  }

  /**
   * This method will be used to check whether any more element is present or
   * not
   *
   * @return more element is present
   */
  public boolean hasNext() {
    return this.fileCounter > 0;
  }

  public void clear() {
    if (null != recordHolderHeapLocal) {
      for (SortTempChunkHolder pageHolder : recordHolderHeapLocal) {
        pageHolder.close();
      }
      recordHolderHeapLocal = null;
    }
  }

  public boolean isStopProcess() {
    return isStopProcess;
  }

  public void setStopProcess(boolean stopProcess) {
    isStopProcess = stopProcess;
  }
}
