/*
 * 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.hadoop.api;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.locks.CarbonLockFactory;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.locks.LockUsage;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatus;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonSessionInfo;
import org.apache.carbondata.core.util.ObjectSerializationUtil;
import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.OperationContext;
import org.apache.carbondata.events.OperationListenerBus;
import org.apache.carbondata.processing.loading.events.LoadEvents;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.util.CarbonLoaderUtil;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobStatus;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.log4j.Logger;

/**
 * OutputCommitter which manages the segments during loading.It commits segment information to the
 * tablestatus file upon success or fail.
 */
public class CarbonOutputCommitter extends FileOutputCommitter {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonOutputCommitter.class.getName());

  private ICarbonLock segmentLock;

  public CarbonOutputCommitter(Path outputPath, TaskAttemptContext context) throws IOException {
    super(outputPath, context);
  }

  /**
   * Update the tablestatus with in-progress while setup the job.
   */
  @Override
  public void setupJob(JobContext context) throws IOException {
    super.setupJob(context);
    boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());
    CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
    if (loadModel.getSegmentId() == null) {
      CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, overwriteSet);
    }
    // Take segment lock
    segmentLock = CarbonLockFactory.getCarbonLockObj(
        loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier(),
        CarbonTablePath.addSegmentPrefix(loadModel.getSegmentId()) + LockUsage.LOCK);
    if (!segmentLock.lockWithRetries()) {
      throw new RuntimeException("Already segment is locked for loading, not supposed happen");
    }
    CarbonTableOutputFormat.setLoadModel(context.getConfiguration(), loadModel);
  }

  @Override
  public void setupTask(TaskAttemptContext context) throws IOException {
    super.setupTask(context);
  }

  /**
   * Update the tablestatus as success after job is success
   */
  @Override
  public void commitJob(JobContext context) throws IOException {
    // comma separated partitions
    String partitionPath = context.getConfiguration().get("carbon.output.partitions.name");
    long t1 = System.currentTimeMillis();
    try {
      super.commitJob(context);
    } catch (IOException e) {
      // ignore, in case of concurrent load it try to remove temporary folders by other load may
      // cause file not found exception. This will not impact carbon load,
      LOGGER.warn(e.getMessage());
    }
    LOGGER.info(
        "$$$ Time taken for the super.commitJob in ms: " + (System.currentTimeMillis() - t1));

    boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());
    CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
    List<PartitionSpec> currentPartitionsOfTable = (List<PartitionSpec>) ObjectSerializationUtil
        .convertStringToObject(context.getConfiguration().get("carbon.currentpartition"));
    if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isHivePartitionTable()) {
      try {
        commitJobForPartition(context, overwriteSet, loadModel, partitionPath,
            currentPartitionsOfTable);
      } catch (Exception e) {
        CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
        LOGGER.error("commit job failed", e);
        throw new IOException(e.getMessage());
      } finally {
        if (segmentLock != null) {
          segmentLock.unlock();
        }
      }
      return;
    }
    LoadMetadataDetails newMetaEntry = loadModel.getCurrentLoadMetadataDetail();
    String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
        + CarbonCommonConstants.FILE_SEPARATOR
        + loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp";
    // Merge all partition files into a single file.
    String segmentFileName = SegmentFileStore.genSegmentFileName(
        loadModel.getSegmentId(), String.valueOf(loadModel.getFactTimeStamp()));
    SegmentFileStore.SegmentFile segmentFile = SegmentFileStore
        .mergeSegmentFiles(readPath, segmentFileName,
            CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath()));
    if (segmentFile != null) {
      if (null == newMetaEntry) {
        throw new RuntimeException("Internal Error");
      }
      // Move all files from temp directory of each segment to partition directory
      SegmentFileStore.moveFromTempFolder(segmentFile,
          loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp",
          loadModel.getTablePath());
      newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
    }
    OperationContext operationContext = (OperationContext) getOperationContext();
    CarbonTable carbonTable = loadModel.getCarbonDataLoadSchema().getCarbonTable();
    String uuid = "";
    newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
    CarbonLoaderUtil
        .populateNewLoadMetaEntry(newMetaEntry, SegmentStatus.SUCCESS, loadModel.getFactTimeStamp(),
            true);
    long segmentSize = CarbonLoaderUtil
        .addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId(), carbonTable);
    if (segmentSize > 0 || overwriteSet) {
      if (operationContext != null) {
        operationContext
            .setProperty(CarbonCommonConstants.CURRENT_SEGMENTFILE, newMetaEntry.getSegmentFile());
        LoadEvents.LoadTablePreStatusUpdateEvent event =
            new LoadEvents.LoadTablePreStatusUpdateEvent(carbonTable.getCarbonTableIdentifier(),
                loadModel);
        try {
          OperationListenerBus.getInstance().fireEvent(event, operationContext);
        } catch (Exception e) {
          throw new IOException(e);
        }
      }
      // After merging index, update newMetaEntry with updated merge index size
      boolean isMergeIndexEnabled = Boolean.parseBoolean(CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT,
              CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT));
      if (isMergeIndexEnabled) {
        CarbonLoaderUtil
            .addIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId(), carbonTable);
      }
      String uniqueId = null;
      if (overwriteSet) {
        if (!loadModel.isCarbonTransactionalTable()) {
          CarbonLoaderUtil.deleteNonTransactionalTableForInsertOverwrite(loadModel);
        } else {
          if (segmentSize == 0) {
            newMetaEntry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
          }
          List<String> partitionList = (List<String>) ObjectSerializationUtil
              .convertStringToObject(partitionPath);
          uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid, partitionList,
              currentPartitionsOfTable);
        }
      } else {
        CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid, false);
      }
      commitJobFinal(context, loadModel, operationContext, carbonTable, uniqueId);
    } else {
      CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
    }
    if (segmentLock != null) {
      segmentLock.unlock();
    }
  }

  private void commitJobFinal(JobContext context, CarbonLoadModel loadModel,
      OperationContext operationContext, CarbonTable carbonTable, String uniqueId)
      throws IOException {
    if (operationContext != null) {
      LoadEvents.LoadTablePostStatusUpdateEvent postStatusUpdateEvent =
          new LoadEvents.LoadTablePostStatusUpdateEvent(loadModel);
      try {
        OperationListenerBus.getInstance()
            .fireEvent(postStatusUpdateEvent, operationContext);
      } catch (Exception e) {
        throw new IOException(e);
      }
    }
    String updateTime =
        context.getConfiguration().get(CarbonTableOutputFormat.UPDATE_TIMESTAMP, uniqueId);
    String segmentsToBeDeleted =
        context.getConfiguration().get(CarbonTableOutputFormat.SEGMENTS_TO_BE_DELETED, "");
    List<Segment> segmentDeleteList = Collections.emptyList();
    if (!segmentsToBeDeleted.trim().isEmpty()) {
      segmentDeleteList = Segment.toSegmentList(segmentsToBeDeleted.split(","), null);
    }
    boolean isUpdateStatusFileUpdateRequired =
        (context.getConfiguration().get(CarbonTableOutputFormat.UPDATE_TIMESTAMP) != null);
    if (updateTime != null) {
      CarbonUpdateUtil.updateTableMetadataStatus(Collections.singleton(loadModel.getSegment()),
          carbonTable, updateTime, true, isUpdateStatusFileUpdateRequired,
          segmentDeleteList, carbonTable.getTableStatusVersion());
    }
  }

  /**
   * re-factory commitJob flow for partition table
   */
  private void commitJobForPartition(JobContext context, boolean overwriteSet,
      CarbonLoadModel loadModel, String partitionPath, List<PartitionSpec> currentPartitionsOfTable)
      throws IOException {
    String size = context.getConfiguration().get("carbon.datasize", "");
    String indexSize = context.getConfiguration().get("carbon.indexsize", "");
    if (size.equalsIgnoreCase("0") || indexSize.equalsIgnoreCase("0")) {
      CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
      return;
    }
    LoadMetadataDetails newMetaEntry = loadModel.getCurrentLoadMetadataDetail();
    CarbonLoaderUtil
        .populateNewLoadMetaEntry(newMetaEntry, SegmentStatus.SUCCESS, loadModel.getFactTimeStamp(),
            true);
    OperationContext operationContext = (OperationContext) getOperationContext();
    CarbonTable carbonTable = loadModel.getCarbonDataLoadSchema().getCarbonTable();
    String uuid = "";
    if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isMV()
        && operationContext != null) {
      uuid = operationContext.getProperty("uuid").toString();
    }
    String tempFolderPath = loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp";
    boolean isMergeIndex = Boolean.parseBoolean(CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT,
            CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT));
    if (!isMergeIndex) {
      // By default carbon.merge.index.in.segment is true and this code will be used for
      // developer debugging purpose.
      Map<String, Set<String>> indexFileNameMap = (Map<String, Set<String>>) ObjectSerializationUtil
          .convertStringToObject(context.getConfiguration().get("carbon.index.files.name"));
      List<String> partitionList =
          (List<String>) ObjectSerializationUtil.convertStringToObject(partitionPath);
      SegmentFileStore.writeSegmentFile(loadModel.getTablePath(), loadModel.getSegmentId(),
          String.valueOf(loadModel.getFactTimeStamp()), partitionList, indexFileNameMap);
      tempFolderPath = null;
    }
    if (operationContext != null) {
      operationContext.setProperty("partitionPath", partitionPath);
      operationContext.setProperty("tempPath", tempFolderPath);
      operationContext.setProperty(
          "carbon.currentpartition",
          context.getConfiguration().get("carbon.currentpartition"));
      LoadEvents.LoadTablePreStatusUpdateEvent event =
          new LoadEvents.LoadTablePreStatusUpdateEvent(carbonTable.getCarbonTableIdentifier(),
              loadModel);
      try {
        OperationListenerBus.getInstance().fireEvent(event, operationContext);
      } catch (Exception e) {
        throw new IOException(e);
      }
    }
    String segmentFileName = SegmentFileStore.genSegmentFileName(
        loadModel.getSegmentId(), String.valueOf(loadModel.getFactTimeStamp()));
    newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
    if (isMergeIndex) {
      newMetaEntry.setIndexSize("" + loadModel.getMetrics().getMergeIndexSize());
    } else if (!StringUtils.isEmpty(indexSize)) {
      newMetaEntry.setIndexSize(indexSize);
    }
    if (!StringUtils.isEmpty(size)) {
      newMetaEntry.setDataSize(size);
    }
    String uniqueId = null;
    if (overwriteSet) {
      List<String> partitionList =
          (List<String>) ObjectSerializationUtil.convertStringToObject(partitionPath);
      uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid, partitionList,
          currentPartitionsOfTable);
    } else {
      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid, false);
    }
    if (operationContext != null) {
      operationContext
          .setProperty(CarbonCommonConstants.CURRENT_SEGMENTFILE, newMetaEntry.getSegmentFile());
    }
    commitJobFinal(context, loadModel, operationContext, carbonTable, uniqueId);
  }

  /**
   * Overwrite the partitions in case of overwrite query. It just updates the partition map files
   * of all segment files.
   */
  private String overwritePartitions(CarbonLoadModel loadModel, LoadMetadataDetails newMetaEntry,
      String uuid, List<String> partitionList, List<PartitionSpec> currentPartitionsOfTable)
      throws IOException {
    CarbonTable table = loadModel.getCarbonDataLoadSchema().getCarbonTable();
    if (partitionList != null && partitionList.size() > 0) {
      // check if any partitions overlaps
      List<String> overlappingPartitions = currentPartitionsOfTable.stream()
          .map(partitionSpec -> partitionSpec.getLocation().toString())
          .filter(partitionList::contains).collect(Collectors.toList());
      if (!overlappingPartitions.isEmpty()) {
        List<LoadMetadataDetails> validLoadMetadataDetails =
            loadModel.getLoadMetadataDetails().stream().filter(loadMetadataDetail ->
                !loadMetadataDetail.getLoadName().equalsIgnoreCase(newMetaEntry.getLoadName()) && (
                    loadMetadataDetail.getSegmentStatus().equals(SegmentStatus.SUCCESS)
                        || loadMetadataDetail.getSegmentStatus()
                        .equals(SegmentStatus.LOAD_PARTIAL_SUCCESS))).collect(Collectors.toList());
        String uniqueId = String.valueOf(System.currentTimeMillis());
        List<String> toBeUpdatedSegments = new ArrayList<>(validLoadMetadataDetails.size());
        List<String> toBeDeletedSegments = new ArrayList<>(validLoadMetadataDetails.size());
        // First drop the partitions from partition mapper files of each segment
        for (LoadMetadataDetails loadMetadataDetail : validLoadMetadataDetails) {
          new SegmentFileStore(table.getTablePath(), loadMetadataDetail.getSegmentFile())
              .dropPartitions(loadMetadataDetail.getLoadName(), partitionList, uniqueId,
                  toBeDeletedSegments, toBeUpdatedSegments);
        }
        newMetaEntry.setUpdateStatusFileName(uniqueId);
        // Commit the removed partitions in carbon store.
        CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid,
            Segment.toSegmentList(toBeDeletedSegments, null),
            Segment.toSegmentList(toBeUpdatedSegments, null), false);
        return uniqueId;
      } else {
        CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid, false);
        return null;
      }
    }
    return null;
  }

  private Object getOperationContext() {
    // when validate segments is disabled in thread local update it to CarbonTableInputFormat
    CarbonSessionInfo carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo();
    if (carbonSessionInfo != null) {
      return carbonSessionInfo.getThreadParams().getExtraInfo("partition.operationcontext");
    }
    return null;
  }

  /**
   * Update the tablestatus as fail if any fail happens.And also clean up the temp folders if any
   * are existed.
   */
  @Override
  public void abortJob(JobContext context, JobStatus.State state) throws IOException {
    try {
      super.abortJob(context, state);
      CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
      CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
      String segmentFileName = loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp();
      LoadMetadataDetails metadataDetail = loadModel.getCurrentLoadMetadataDetail();
      if (metadataDetail != null) {
        // In case the segment file is already created for this job then just link it so that it
        // will be used while cleaning.
        if (!metadataDetail.getSegmentStatus().equals(SegmentStatus.SUCCESS)) {
          String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
              + CarbonTablePath.SEGMENT_EXT;
          if (FileFactory.getCarbonFile(readPath).exists()) {
            metadataDetail.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
          }
        }
      }
      // Clean the temp files
      CarbonFile segTmpFolder = FileFactory.getCarbonFile(
          CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName + ".tmp");
      // delete temp segment folder
      if (segTmpFolder.exists()) {
        FileFactory.deleteAllCarbonFilesOfDir(segTmpFolder);
      }
      CarbonFile segmentFilePath = FileFactory.getCarbonFile(
          CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
              + CarbonTablePath.SEGMENT_EXT);
      // Delete the temp data folders of this job if exists
      if (segmentFilePath.exists()) {
        SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),
            segmentFileName + CarbonTablePath.SEGMENT_EXT);
        SegmentFileStore.removeTempFolder(fileStore.getLocationMap(), segmentFileName + ".tmp",
            loadModel.getTablePath());
      }
      LOGGER.error("Loading failed with job status : " + state);
    } finally {
      if (segmentLock != null) {
        segmentLock.unlock();
      }
    }
  }

}
