/*
 * 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.core.mutate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonCommonConstantsInternal;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.mutate.data.BlockMappingVO;
import org.apache.carbondata.core.mutate.data.RowCountDetailsVO;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatus;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * This class contains all update utility methods
 */
public class CarbonUpdateUtil {

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

  /**
   * returns required filed from tuple id
   *
   */
  public static String getRequiredFieldFromTID(String Tid, int index) {
    return Tid.split(CarbonCommonConstants.FILE_SEPARATOR)[index];
  }

  /**
   * returns required filed from tuple id
   *
   * @param Tid
   * @param tid
   * @return
   */
  public static String getRequiredFieldFromTID(String Tid, TupleIdEnum tid) {
    return Tid.split("/")[tid.getTupleIdIndex()];
  }

  /**
   * returns segment along with block id
   * @param Tid
   * @return
   */
  public static String getSegmentWithBlockFromTID(String Tid, boolean isPartitionTable) {
    if (isPartitionTable) {
      return getRequiredFieldFromTID(Tid, TupleIdEnum.SEGMENT_ID);
    }
    return getRequiredFieldFromTID(Tid, TupleIdEnum.SEGMENT_ID)
        + CarbonCommonConstants.FILE_SEPARATOR + getRequiredFieldFromTID(Tid, TupleIdEnum.BLOCK_ID);
  }

  /**
   * Returns block path from tuple id
   */
  public static String getTableBlockPath(String tid, String tablePath, boolean isStandardTable) {
    String partField = getRequiredFieldFromTID(tid, TupleIdEnum.PART_ID);
    // If it has segment file then part field can be appended directly to table path
    if (!isStandardTable) {
      return tablePath + CarbonCommonConstants.FILE_SEPARATOR + partField.replace("#", "/");
    }
    String part = CarbonTablePath.addPartPrefix(partField);
    String segment =
            CarbonTablePath.addSegmentPrefix(getRequiredFieldFromTID(tid, TupleIdEnum.SEGMENT_ID));
    return CarbonTablePath.getFactDir(tablePath) + CarbonCommonConstants.FILE_SEPARATOR + part
            + CarbonCommonConstants.FILE_SEPARATOR + segment;
  }

  /**
   * returns delete delta file path
   *
   * @param blockPath
   * @param blockPath
   * @param timestamp
   * @return
   */
  public static String getDeleteDeltaFilePath(String blockPath, String blockName,
                                              String timestamp) {
    return blockPath + CarbonCommonConstants.FILE_SEPARATOR + blockName
        + CarbonCommonConstants.HYPHEN + timestamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT;

  }

  /**
   * @param updateDetailsList
   * @param table
   * @param updateStatusFileIdentifier
   * @return
   */
  public static boolean updateSegmentStatus(List<SegmentUpdateDetails> updateDetailsList,
      CarbonTable table, String updateStatusFileIdentifier, boolean isCompaction) {
    boolean status = false;
    SegmentUpdateStatusManager segmentUpdateStatusManager = new SegmentUpdateStatusManager(table);
    ICarbonLock updateLock = segmentUpdateStatusManager.getTableUpdateStatusLock();
    boolean lockStatus = false;

    try {
      lockStatus = updateLock.lockWithRetries();
      if (lockStatus) {

        // read the existing file if present and update the same.
        SegmentUpdateDetails[] oldDetails = segmentUpdateStatusManager
                .getUpdateStatusDetails();

        List<SegmentUpdateDetails> oldList = new ArrayList(Arrays.asList(oldDetails));

        for (SegmentUpdateDetails newBlockEntry : updateDetailsList) {
          mergeSegmentUpdate(isCompaction, oldList, newBlockEntry);
        }

        List<SegmentUpdateDetails> updateDetailsValidSeg = new ArrayList<>();
        Set<String> loadDetailsSet = new HashSet<>();
        for (LoadMetadataDetails details : segmentUpdateStatusManager.getLoadMetadataDetails()) {
          loadDetailsSet.add(details.getLoadName());
        }
        for (SegmentUpdateDetails updateDetails : oldList) {
          if (loadDetailsSet.contains(updateDetails.getSegmentName())) {
            // we should only keep the update info of segments in table status, especially after
            // compaction and clean files some compacted segments will be removed. It can keep
            // tableupdatestatus file in small size which is good for performance.
            updateDetailsValidSeg.add(updateDetails);
          }
        }
        segmentUpdateStatusManager
            .writeLoadDetailsIntoFile(updateDetailsValidSeg, updateStatusFileIdentifier);
        status = true;
      } else {
        LOGGER.error("Not able to acquire the segment update lock.");
        status = false;
      }
    } catch (IOException e) {
      status = false;
    } finally {
      if (lockStatus) {
        if (updateLock.unlock()) {
          LOGGER.info("Unlock the segment update lock successful.");
        } else {
          LOGGER.error("Not able to unlock the segment update lock.");
        }
      }
    }
    return status;
  }

  public static void mergeSegmentUpdate(boolean isCompaction, List<SegmentUpdateDetails> oldList,
      SegmentUpdateDetails newBlockEntry) {
    int index = oldList.indexOf(newBlockEntry);
    if (index != -1) {
      // update the element in existing list.
      SegmentUpdateDetails blockDetail = oldList.get(index);
      if (blockDetail.getDeleteDeltaStartTimestamp().isEmpty() || isCompaction) {
        blockDetail
            .setDeleteDeltaStartTimestamp(newBlockEntry.getDeleteDeltaStartTimestamp());
      }
      blockDetail.setDeleteDeltaEndTimestamp(newBlockEntry.getDeleteDeltaEndTimestamp());
      blockDetail.setSegmentStatus(newBlockEntry.getSegmentStatus());
      blockDetail.setDeletedRowsInBlock(newBlockEntry.getDeletedRowsInBlock());
      // If the start and end time is different then the delta is there in multiple files so
      // add them to the list to get the delta files easily with out listing.
      if (!blockDetail.getDeleteDeltaStartTimestamp()
          .equals(blockDetail.getDeleteDeltaEndTimestamp())) {
        blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaStartTimestamp());
        blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaEndTimestamp());
      } else {
        blockDetail.setDeltaFileStamps(null);
      }
    } else {
      // add the new details to the list.
      oldList.add(newBlockEntry);
    }
  }

  /**
   * Update table status
   * @param updatedSegmentsList
   * @param table
   * @param updatedTimeStamp
   * @param isTimestampUpdateRequired
   * @param segmentsToBeDeleted
   * @return
   */
  public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
      CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
      List<Segment> segmentsToBeDeleted) {
    return updateTableMetadataStatus(updatedSegmentsList, table, updatedTimeStamp,
        isTimestampUpdateRequired, segmentsToBeDeleted, new ArrayList<Segment>(), "");
  }

  /**
   *
   * @param updatedSegmentsList
   * @param table
   * @param updatedTimeStamp
   * @param isTimestampUpdateRequired
   * @param segmentsToBeDeleted
   * @return
   */
  public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
      CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
      List<Segment> segmentsToBeDeleted, List<Segment> segmentFilesTobeUpdated, String uuid) {

    boolean status = false;
    String metaDataFilepath = table.getMetadataPath();
    AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
    String tableStatusPath =
        CarbonTablePath.getTableStatusFilePathWithUUID(identifier.getTablePath(), uuid);
    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(identifier);

    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
    boolean lockStatus = false;
    try {
      lockStatus = carbonLock.lockWithRetries();
      if (lockStatus) {
        LOGGER.info(
                "Acquired lock for table" + table.getDatabaseName() + "." + table.getTableName()
                        + " for table status update");

        LoadMetadataDetails[] listOfLoadFolderDetailsArray =
                SegmentStatusManager.readLoadMetadata(metaDataFilepath);

        for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {

          if (isTimestampUpdateRequired) {
            // we are storing the link between the 2 status files in the segment 0 only.
            if (loadMetadata.getLoadName().equalsIgnoreCase("0")) {
              loadMetadata.setUpdateStatusFileName(
                      CarbonUpdateUtil.getUpdateStatusFileName(updatedTimeStamp));
            }

            // if the segments is in the list of marked for delete then update the status.
            if (segmentsToBeDeleted.contains(new Segment(loadMetadata.getLoadName()))) {
              loadMetadata.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
              loadMetadata.setModificationOrDeletionTimestamp(Long.parseLong(updatedTimeStamp));
            }
          }
          for (Segment segName : updatedSegmentsList) {
            if (loadMetadata.getLoadName().equalsIgnoreCase(segName.getSegmentNo())) {
              // if this call is coming from the delete delta flow then the time stamp
              // String will come empty then no need to write into table status file.
              if (isTimestampUpdateRequired) {
                // if in case of update flow.
                if (loadMetadata.getUpdateDeltaStartTimestamp().isEmpty()) {
                  // this means for first time it is getting updated .
                  loadMetadata.setUpdateDeltaStartTimestamp(updatedTimeStamp);
                }
                // update end timestamp for each time.
                loadMetadata.setUpdateDeltaEndTimestamp(updatedTimeStamp);
              }
              if (segmentFilesTobeUpdated
                  .contains(Segment.toSegment(loadMetadata.getLoadName(), null))) {
                loadMetadata.setSegmentFile(loadMetadata.getLoadName() + "_" + updatedTimeStamp
                    + CarbonTablePath.SEGMENT_EXT);
              }
            }
          }
        }

        try {
          SegmentStatusManager
                  .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray);
        } catch (IOException e) {
          return false;
        }

        status = true;
      } else {
        LOGGER.error("Not able to acquire the lock for Table status update for table " + table
                .getDatabaseName() + "." + table.getTableName());
      }
    } finally {
      if (lockStatus) {
        if (carbonLock.unlock()) {
          LOGGER.info(
                 "Table unlocked successfully after table status update" + table.getDatabaseName()
                          + "." + table.getTableName());
        } else {
          LOGGER.error(
                  "Unable to unlock Table lock for table" + table.getDatabaseName() + "." + table
                          .getTableName() + " during table status update");
        }
      }
    }
    return status;

  }

  /**
   * gets the file name of the update status file. by appending the latest timestamp to it.
   *
   * @param updatedTimeStamp
   * @return
   */
  public static String getUpdateStatusFileName(String updatedTimeStamp) {
    return CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME + CarbonCommonConstants.HYPHEN
            + updatedTimeStamp;
  }

  /**
   * This will handle the clean up cases if the update fails.
   *
   * @param table
   * @param timeStamp
   */
  public static void cleanStaleDeltaFiles(CarbonTable table, final String timeStamp) {

    AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
    String partitionDir = CarbonTablePath.getPartitionDir(identifier.getTablePath());
    CarbonFile file =
            FileFactory.getCarbonFile(partitionDir);
    if (!file.exists()) {
      return;
    }
    for (CarbonFile eachDir : file.listFiles()) {
      // for each dir check if the file with the delta timestamp is present or not.
      CarbonFile[] toBeDeleted = eachDir.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          String fileName = file.getName();
          return (fileName.endsWith(timeStamp + CarbonCommonConstants.UPDATE_DELTA_FILE_EXT)
                  || fileName.endsWith(timeStamp + CarbonCommonConstants.UPDATE_INDEX_FILE_EXT)
                  || fileName.endsWith(timeStamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT));
        }
      });
      // deleting the files of a segment.
      try {
        CarbonUtil.deleteFoldersAndFilesSilent(toBeDeleted);
      } catch (IOException e) {
        LOGGER.error("Exception in deleting the delta files." + e);
      } catch (InterruptedException e) {
        LOGGER.error("Exception in deleting the delta files." + e);
      }
    }
  }

  /**
   * returns timestamp as long value
   *
   * @param timestamp
   * @return
   */
  public static Long getTimeStampAsLong(String timestamp) {
    try {
      return Long.parseLong(timestamp);
    } catch (NumberFormatException nfe) {
      String errorMsg = "Invalid timestamp : " + timestamp;
      LOGGER.error(errorMsg);
      return null;
    }
  }

  /**
   * returns integer value from given string
   *
   * @param value
   * @return
   * @throws Exception
   */
  public static Integer getIntegerValue(String value) throws Exception {
    try {
      return Integer.parseInt(value);
    } catch (NumberFormatException nfe) {
      LOGGER.error("Invalid row : " + value + nfe.getLocalizedMessage());
      throw new Exception("Invalid row : " + nfe.getLocalizedMessage());
    }
  }

  /**
   * return only block name from completeBlockName
   *
   * @param completeBlockName
   * @return
   */
  public static String getBlockName(String completeBlockName) {
    return completeBlockName
        .substring(0, completeBlockName.lastIndexOf(CarbonCommonConstants.HYPHEN));
  }

  /**
   * returns segment id from segment name
   *
   * @param segmentName
   * @return
   */
  public static String getSegmentId(String segmentName) {
    return segmentName.split(CarbonCommonConstants.UNDERSCORE)[1];
  }

  public static long getLatestTaskIdForSegment(Segment segment, String tablePath)
      throws IOException {
    long max = 0;
    List<String> dataFiles = new ArrayList<>();
    if (segment.getSegmentFileName() != null) {
      SegmentFileStore fileStore = new SegmentFileStore(tablePath, segment.getSegmentFileName());
      fileStore.readIndexFiles(FileFactory.getConfiguration());
      Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
      List<String> dataFilePaths = new ArrayList<>();
      for (List<String> paths : indexFilesMap.values()) {
        dataFilePaths.addAll(paths);
      }
      for (String dataFilePath : dataFilePaths) {
        dataFiles.add(new Path(dataFilePath).getName());
      }

    } else {
      String segmentDirPath = CarbonTablePath.getSegmentPath(tablePath, segment.getSegmentNo());
      // scan all the carbondata files and get the latest task ID.
      CarbonFile segmentDir =
          FileFactory.getCarbonFile(segmentDirPath);
      CarbonFile[] carbonDataFiles = segmentDir.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {

          if (file.getName().endsWith(CarbonCommonConstants.FACT_FILE_EXT)) {
            return true;
          }
          return false;
        }
      });
      for (CarbonFile carbonDataFile : carbonDataFiles) {
        dataFiles.add(carbonDataFile.getName());
      }
    }
    for (String name : dataFiles) {
      long taskNumber =
          Long.parseLong(CarbonTablePath.DataFileUtil.getTaskNo(name).split("_")[0]);
      if (taskNumber > max) {
        max = taskNumber;
      }
    }
    // return max task No
    return max;

  }

  /**
   * Handling of the clean up of old carbondata files, index files , delete delta,
   * update status files.
   * @param table clean up will be handled on this table.
   * @param forceDelete if true then max query execution timeout will not be considered.
   */
  public static void cleanUpDeltaFiles(CarbonTable table, boolean forceDelete) throws IOException {

    SegmentStatusManager ssm = new SegmentStatusManager(table.getAbsoluteTableIdentifier());

    LoadMetadataDetails[] details =
        SegmentStatusManager.readLoadMetadata(table.getMetadataPath());

    SegmentUpdateStatusManager updateStatusManager = new SegmentUpdateStatusManager(table);
    SegmentUpdateDetails[] segmentUpdateDetails = updateStatusManager.getUpdateStatusDetails();
    // hold all the segments updated so that wen can check the delta files in them, ne need to
    // check the others.
    Set<String> updatedSegments = new HashSet<>();
    for (SegmentUpdateDetails updateDetails : segmentUpdateDetails) {
      updatedSegments.add(updateDetails.getSegmentName());
    }

    String validUpdateStatusFile = "";

    boolean isAbortedFile = true;

    boolean isInvalidFile = false;

    List<Segment> segmentFilesToBeUpdated = new ArrayList<>();

    // take the update status file name from 0th segment.
    validUpdateStatusFile = ssm.getUpdateStatusFileName(details);
    // scan through each segment.
    for (LoadMetadataDetails segment : details) {
      // if this segment is valid then only we will go for delta file deletion.
      // if the segment is mark for delete or compacted then any way it will get deleted.
      if (segment.getSegmentStatus() == SegmentStatus.SUCCESS
              || segment.getSegmentStatus() == SegmentStatus.LOAD_PARTIAL_SUCCESS) {
        // when there is no update operations done on table, then no need to go ahead. So
        // just check the update delta start timestamp and proceed if not empty
        if (!segment.getUpdateDeltaStartTimestamp().isEmpty()
                || updatedSegments.contains(segment.getLoadName())) {
          // take the list of files from this segment.
          String segmentPath = CarbonTablePath.getSegmentPath(
              table.getAbsoluteTableIdentifier().getTablePath(), segment.getLoadName());
          CarbonFile segDir =
              FileFactory.getCarbonFile(segmentPath);
          CarbonFile[] allSegmentFiles = segDir.listFiles();

          // scan through the segment and find the carbon data files and index files.
          boolean updateSegmentFile = false;
          // deleting of the aborted file scenario.
          if (deleteStaleCarbonDataFiles(segment, allSegmentFiles, updateStatusManager)) {
            updateSegmentFile = true;
          }

          // get Invalid update  delta files.
          CarbonFile[] invalidUpdateDeltaFiles = updateStatusManager
              .getUpdateDeltaFilesList(segment, false,
                  CarbonCommonConstants.UPDATE_DELTA_FILE_EXT, true, allSegmentFiles,
                  isInvalidFile);

          // now for each invalid delta file need to check the query execution time out
          // and then delete.
          for (CarbonFile invalidFile : invalidUpdateDeltaFiles) {
            compareTimestampsAndDelete(invalidFile, forceDelete, false);
          }
          // do the same for the index files.
          CarbonFile[] invalidIndexFiles = updateStatusManager
              .getUpdateDeltaFilesList(segment, false,
                  CarbonCommonConstants.UPDATE_INDEX_FILE_EXT, true, allSegmentFiles,
                  isInvalidFile);

          // now for each invalid index file need to check the query execution time out
          // and then delete.

          for (CarbonFile invalidFile : invalidIndexFiles) {
            if (compareTimestampsAndDelete(invalidFile, forceDelete, false)) {
              updateSegmentFile = true;
            }
          }
          // now handle all the delete delta files which needs to be deleted.
          // there are 2 cases here .
          // 1. if the block is marked as compacted then the corresponding delta files
          //    can be deleted if query exec timeout is done.
          // 2. if the block is in success state then also there can be delete
          //    delta compaction happened and old files can be deleted.

          SegmentUpdateDetails[] updateDetails = updateStatusManager.readLoadMetadata();
          for (SegmentUpdateDetails block : updateDetails) {
            CarbonFile[] completeListOfDeleteDeltaFiles;
            CarbonFile[] invalidDeleteDeltaFiles;

            if (!block.getSegmentName().equalsIgnoreCase(segment.getLoadName())) {
              continue;
            }

            // aborted scenario.
            invalidDeleteDeltaFiles = updateStatusManager
                .getDeleteDeltaInvalidFilesList(block, false,
                    allSegmentFiles, isAbortedFile);
            for (CarbonFile invalidFile : invalidDeleteDeltaFiles) {
              boolean doForceDelete = true;
              compareTimestampsAndDelete(invalidFile, doForceDelete, false);
            }

            // case 1
            if (CarbonUpdateUtil.isBlockInvalid(block.getSegmentStatus())) {
              completeListOfDeleteDeltaFiles = updateStatusManager
                  .getDeleteDeltaInvalidFilesList(block, true,
                      allSegmentFiles, isInvalidFile);
              for (CarbonFile invalidFile : completeListOfDeleteDeltaFiles) {
                compareTimestampsAndDelete(invalidFile, forceDelete, false);
              }

            } else {
              invalidDeleteDeltaFiles = updateStatusManager
                  .getDeleteDeltaInvalidFilesList(block, false,
                      allSegmentFiles, isInvalidFile);
              for (CarbonFile invalidFile : invalidDeleteDeltaFiles) {
                compareTimestampsAndDelete(invalidFile, forceDelete, false);
              }
            }
          }
          if (updateSegmentFile) {
            segmentFilesToBeUpdated.add(
                new Segment(segment.getLoadName(), segment.getSegmentFile(), null));
          }
        }
        // handle cleanup of merge index files and data files after small files merge happened for
        // SI table
        cleanUpDataFilesAfterSmallFilesMergeForSI(table, segment);
      }
    }
    String UUID = String.valueOf(System.currentTimeMillis());
    List<Segment> segmentFilesToBeUpdatedLatest = new ArrayList<>();
    CarbonFile segmentFilesLocation =
        FileFactory.getCarbonFile(CarbonTablePath.getSegmentFilesLocation(table.getTablePath()));
    for (Segment segment : segmentFilesToBeUpdated) {
      SegmentFileStore fileStore =
          new SegmentFileStore(table.getTablePath(), segment.getSegmentFileName());
      segment.setSegmentMetaDataInfo(fileStore.getSegmentFile().getSegmentMetaDataInfo());
      String updatedSegmentFile = SegmentFileStore
          .writeSegmentFile(table, segment.getSegmentNo(), UUID,
              CarbonTablePath.getSegmentPath(table.getTablePath(), segment.getSegmentNo()),
              segment.getSegmentMetaDataInfo());
      segmentFilesToBeUpdatedLatest.add(new Segment(segment.getSegmentNo(), updatedSegmentFile));

      // delete the old segment files
      CarbonFile[] invalidSegmentFiles = segmentFilesLocation.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          return !file.getName().equalsIgnoreCase(updatedSegmentFile);
        }
      });
      for (CarbonFile invalidSegmentFile : invalidSegmentFiles) {
        invalidSegmentFile.delete();
      }
    }
    if (segmentFilesToBeUpdated.size() > 0) {
      updateTableMetadataStatus(
          new HashSet<Segment>(segmentFilesToBeUpdated),
          table,
          UUID,
          false,
          new ArrayList<Segment>(),
          segmentFilesToBeUpdatedLatest, "");
    }

    // delete the update table status files which are old.
    if (null != validUpdateStatusFile && !validUpdateStatusFile.isEmpty()) {

      final String updateStatusTimestamp = validUpdateStatusFile
          .substring(validUpdateStatusFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1);

      String tablePath = table.getAbsoluteTableIdentifier().getTablePath();
      CarbonFile metaFolder = FileFactory.getCarbonFile(
          CarbonTablePath.getMetadataPath(tablePath));

      CarbonFile[] invalidUpdateStatusFiles = metaFolder.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          if (file.getName().startsWith(CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME)) {

            // CHECK if this is valid or not.
            // we only send invalid ones to delete.
            return !file.getName().endsWith(updateStatusTimestamp);
          }
          return false;
        }
      });

      for (CarbonFile invalidFile : invalidUpdateStatusFiles) {

        compareTimestampsAndDelete(invalidFile, forceDelete, true);
      }
    }
  }

  /**
   * this is the clean up added specifically for SI table, because after we merge the data files
   * inside the secondary index table, we need to delete the stale carbondata files.
   * refer {@link org.apache.spark.sql.secondaryindex.rdd.CarbonSIRebuildRDD}
   */
  private static void cleanUpDataFilesAfterSmallFilesMergeForSI(CarbonTable table,
      LoadMetadataDetails segment) throws IOException {
    if (table.isIndexTable()) {
      String segmentPath = CarbonTablePath
          .getSegmentPath(table.getAbsoluteTableIdentifier().getTablePath(),
              segment.getLoadName());
      CarbonFile segmentDirPath =
          FileFactory.getCarbonFile(segmentPath);
      CarbonFile[] allFilesOfSegment = segmentDirPath.listFiles();
      long startTimeStampFinal = segment.getLoadStartTime();
      long endTimeStampFinal = segment.getLoadEndTime();
      boolean deleteFile;
      for (CarbonFile file : allFilesOfSegment) {
        deleteFile = false;
        String fileTimestamp =
            CarbonTablePath.DataFileUtil.getTimeStampFromFileName(file.getName());
        // check for old files before load start time and the aborted files after end time
        if ((file.getName().endsWith(CarbonTablePath.CARBON_DATA_EXT) || file.getName()
            .endsWith(CarbonTablePath.INDEX_FILE_EXT)) && (
            Long.parseLong(fileTimestamp) < startTimeStampFinal
                || Long.parseLong(fileTimestamp) > endTimeStampFinal)) {
          deleteFile = true;
        } else if (file.getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)
            && Long.parseLong(fileTimestamp) < startTimeStampFinal) {
          deleteFile = true;
        }
        if (deleteFile) {
          // delete the files and folders.
          try {
            LOGGER.info("Deleting the invalid file : " + file.getName());
            CarbonUtil.deleteFoldersAndFiles(file);
          } catch (IOException e) {
            LOGGER.error("Error in clean up of merged files." + e.getMessage(), e);
          } catch (InterruptedException e) {
            LOGGER.error("Error in clean up of merged files." + e.getMessage(), e);
          }
        }
      }
    }
  }

  /**
   * This function deletes all the stale carbondata files during clean up before update operation
   * one scenario is if update operation is abruptly stopped before update of table status then
   * the carbondata file created during update operation is stale file and it will be deleted in
   * this function in next update operation
   * @param segment
   * @param allSegmentFiles
   * @param updateStatusManager
   */
  private static boolean deleteStaleCarbonDataFiles(LoadMetadataDetails segment,
      CarbonFile[] allSegmentFiles, SegmentUpdateStatusManager updateStatusManager) {
    CarbonFile[] invalidUpdateDeltaFiles = updateStatusManager
        .getUpdateDeltaFilesList(segment, false,
            CarbonCommonConstants.UPDATE_DELTA_FILE_EXT, true, allSegmentFiles,
            true);
    // now for each invalid delta file need to check the query execution time out
    // and then delete.
    for (CarbonFile invalidFile : invalidUpdateDeltaFiles) {
      compareTimestampsAndDelete(invalidFile, true, false);
    }
    // do the same for the index files.
    CarbonFile[] invalidIndexFiles = updateStatusManager
        .getUpdateDeltaFilesList(segment, false,
            CarbonCommonConstants.UPDATE_INDEX_FILE_EXT, true, allSegmentFiles,
            true);
    // now for each invalid index file need to check the query execution time out
    // and then delete.
    boolean updateSegmentFile = false;
    for (CarbonFile invalidFile : invalidIndexFiles) {
      if (compareTimestampsAndDelete(invalidFile, true, false)) {
        updateSegmentFile = true;
      }
    }
    return updateSegmentFile;
  }

  /**
   * This will tell whether the max query timeout has been expired or not.
   * @param fileTimestamp
   * @return
   */
  public static boolean isMaxQueryTimeoutExceeded(long fileTimestamp) {
    // record current time.
    long currentTime = CarbonUpdateUtil.readCurrentTime();
    int maxTime;
    try {
      maxTime = Integer.parseInt(CarbonProperties.getInstance()
              .getProperty(CarbonCommonConstants.MAX_QUERY_EXECUTION_TIME));
    } catch (NumberFormatException e) {
      maxTime = CarbonCommonConstants.DEFAULT_MAX_QUERY_EXECUTION_TIME;
    }

    long difference = currentTime - fileTimestamp;

    long minutesElapsed = (difference / (1000 * 60));

    return minutesElapsed > maxTime;

  }

  /**
   *
   * @param invalidFile
   * @param forceDelete
   * @param isUpdateStatusFile if true then the parsing of file name logic changes.
   */
  private static boolean compareTimestampsAndDelete(
      CarbonFile invalidFile,
      boolean forceDelete, boolean isUpdateStatusFile) {
    boolean isDeleted = false;
    Long fileTimestamp;

    if (isUpdateStatusFile) {
      fileTimestamp = CarbonUpdateUtil.getTimeStampAsLong(invalidFile.getName()
              .substring(invalidFile.getName().lastIndexOf(CarbonCommonConstants.HYPHEN) + 1));
    } else {
      fileTimestamp = CarbonUpdateUtil.getTimeStampAsLong(
              CarbonTablePath.DataFileUtil.getTimeStampFromFileName(invalidFile.getName()));
    }

    // This check is because, when there are some invalid files like tableStatusUpdate.write files
    // present in store [[which can happen during delete or update if the disk is full or hdfs quota
    // is finished]] then fileTimestamp will be null, in that case check for max query out and
    // delete the .write file after timeout
    if (fileTimestamp == null) {
      String tableUpdateStatusFilename = invalidFile.getName();
      if (tableUpdateStatusFilename.endsWith(".write")) {
        long tableUpdateStatusFileTimeStamp = Long.parseLong(
            CarbonTablePath.DataFileUtil.getTimeStampFromFileName(tableUpdateStatusFilename));
        if (isMaxQueryTimeoutExceeded(tableUpdateStatusFileTimeStamp)) {
          isDeleted = deleteInvalidFiles(invalidFile);
        }
      }
    } else {
      // if the timestamp of the file is more than the current time by query execution timeout.
      // then delete that file.
      if (CarbonUpdateUtil.isMaxQueryTimeoutExceeded(fileTimestamp) || forceDelete) {
        isDeleted = deleteInvalidFiles(invalidFile);
      }
    }
    return isDeleted;
  }

  private static boolean deleteInvalidFiles(CarbonFile invalidFile) {
    boolean isDeleted;
    try {
      LOGGER.info("deleting the invalid file : " + invalidFile.getName());
      CarbonUtil.deleteFoldersAndFiles(invalidFile);
      isDeleted = true;
    } catch (IOException | InterruptedException e) {
      LOGGER.error("error in clean up of invalid files." + e.getMessage(), e);
      isDeleted = false;
    }
    return isDeleted;
  }

  public static boolean isBlockInvalid(SegmentStatus blockStatus) {
    return blockStatus == SegmentStatus.COMPACTED || blockStatus == SegmentStatus.MARKED_FOR_DELETE;
  }

  /**
   * This will return the current time in millis.
   * @return
   */
  public static long readCurrentTime() {
    return System.currentTimeMillis();
  }

  /**
   *
   * @param details
   * @param segmentBlockCount
   */
  public static void decrementDeletedBlockCount(SegmentUpdateDetails details,
                                                Map<String, Long> segmentBlockCount) {

    String segId = details.getSegmentName();

    segmentBlockCount.put(details.getSegmentName(), segmentBlockCount.get(segId) - 1);

  }

  /**
   *
   * @param segmentBlockCount
   * @return
   */
  public static List<Segment> getListOfSegmentsToMarkDeleted(Map<String, Long> segmentBlockCount) {
    List<Segment> segmentsToBeDeleted =
        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);

    for (Map.Entry<String, Long> eachSeg : segmentBlockCount.entrySet()) {

      if (eachSeg.getValue() == 0) {
        segmentsToBeDeleted.add(new Segment(eachSeg.getKey(), ""));
      }

    }
    return segmentsToBeDeleted;
  }

  /**
   * Return row count of input block
   */
  public static long getRowCount(BlockMappingVO blockMappingVO, CarbonTable carbonTable) {
    if (blockMappingVO.getBlockRowCountMapping().size() == 1
        && blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT)
        != null) {
      return blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT);
    }
    SegmentUpdateStatusManager updateStatusManager =
        new SegmentUpdateStatusManager(carbonTable);
    long rowCount = 0;
    Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping();
    for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) {
      String key = blockRowEntry.getKey();
      long alreadyDeletedCount = 0;
      SegmentUpdateDetails detail = updateStatusManager.getDetailsForABlock(key);
      if (detail != null) {
        alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock());
      }
      rowCount += (blockRowEntry.getValue() - alreadyDeletedCount);
    }
    return rowCount;
  }

  /**
   *
   * @param blockMappingVO
   * @param segmentUpdateStatusManager
   */
  public static void createBlockDetailsMap(BlockMappingVO blockMappingVO,
                                           SegmentUpdateStatusManager segmentUpdateStatusManager) {

    Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping();

    Map<String, RowCountDetailsVO> outputMap =
            new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);

    for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) {
      String key = blockRowEntry.getKey();
      long alreadyDeletedCount = 0;

      SegmentUpdateDetails detail = segmentUpdateStatusManager.getDetailsForABlock(key);

      if (null != detail) {

        alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock());

      }

      RowCountDetailsVO rowCountDetailsVO =
              new RowCountDetailsVO(blockRowEntry.getValue(), alreadyDeletedCount);
      outputMap.put(key, rowCountDetailsVO);

    }

    blockMappingVO.setCompleteBlockRowDetailVO(outputMap);

  }

  /**
   *
   * @param segID
   * @param blockName
   * @return
   */
  public static String getSegmentBlockNameKey(String segID, String blockName,
      boolean isPartitionTable) {
    String blockNameWithOutPart = blockName
        .substring(blockName.indexOf(CarbonCommonConstants.HYPHEN) + 1,
            blockName.lastIndexOf(CarbonTablePath.getCarbonDataExtension()));
    if (isPartitionTable) {
      return blockNameWithOutPart;
    }
    return segID + CarbonCommonConstants.FILE_SEPARATOR + blockNameWithOutPart;
  }

  /**
   * Below method will be used to get the latest delete delta file timestamp
   * @param deleteDeltaFiles
   * @return latest delete delta file time stamp
   */
  public static long getLatestDeleteDeltaTimestamp(String[] deleteDeltaFiles) {
    long latestTimestamp = 0;
    for (int i = 0; i < deleteDeltaFiles.length; i++) {
      long convertTimeStampToLong = Long.parseLong(
          CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(deleteDeltaFiles[i]));
      if (latestTimestamp < convertTimeStampToLong) {
        latestTimestamp = convertTimeStampToLong;
      }
    }
    return latestTimestamp;
  }
}