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

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
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.filesystem.CarbonFileFilter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
import org.apache.carbondata.core.fileoperations.FileWriteOperation;
import org.apache.carbondata.core.index.IndexStoreManager;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.index.TableIndex;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore;
import org.apache.carbondata.core.locks.CarbonLockUtil;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
import org.apache.carbondata.core.segmentmeta.SegmentColumnMetaDataInfo;
import org.apache.carbondata.core.segmentmeta.SegmentMetaDataInfo;
import org.apache.carbondata.core.segmentmeta.SegmentMetaDataInfoStats;
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.CarbonUtil;
import org.apache.carbondata.core.util.DataFileFooterConverter;
import org.apache.carbondata.core.util.ObjectSerializationUtil;
import org.apache.carbondata.core.util.TrashUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import com.google.gson.Gson;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * Provide read and write support for segment file associated with each segment
 */
public class SegmentFileStore {

  private static final Logger LOGGER = LogServiceFactory.getLogService(
      SegmentFileStore.class.getCanonicalName());

  private SegmentFile segmentFile;

  /**
   * Here key folder path and values are index files in it.
   */
  private Map<String, List<String>> indexFilesMap;

  private String tablePath;

  public SegmentFileStore(String tablePath, String segmentFileName) throws IOException {
    this.tablePath = tablePath;
    this.segmentFile = readSegment(tablePath, segmentFileName);
  }

  /**
   * Write segment information to the segment folder with index file name and
   * corresponding partitions.
   */
  public static void writeSegmentFileForPartitionTable(String tablePath, final String taskNo,
      String location, String timeStamp, List<String> partitionNames) throws IOException {
    writeSegmentFileForPartitionTable(tablePath, taskNo, location, timeStamp, partitionNames,
        false);
  }

  /**
   * Method to create and write the segment file, removes the temporary directories from all the
   * respective partition directories. This method is invoked only when {@link
   * CarbonCommonConstants#CARBON_MERGE_INDEX_IN_SEGMENT} is disabled.
   * @param tablePath Table path
   * @param segmentId Segment id
   * @param timeStamp FactTimeStamp
   * @param partitionNames Partition names list
   * @param indexFileNames Index files map with partition as key and index file names set as value
   * @throws IOException
   */
  public static void writeSegmentFile(String tablePath, String segmentId, String timeStamp,
      List<String> partitionNames, Map<String, Set<String>> indexFileNames) throws IOException {
    SegmentFileStore.SegmentFile finalSegmentFile = null;
    boolean isRelativePath;
    String partitionLoc;
    for (String partition : partitionNames) {
      isRelativePath = false;
      partitionLoc = partition;
      if (partitionLoc.startsWith(tablePath)) {
        partitionLoc = partitionLoc.substring(tablePath.length());
        isRelativePath = true;
      }
      SegmentFileStore.SegmentFile segmentFile = new SegmentFileStore.SegmentFile();
      SegmentFileStore.FolderDetails folderDetails = new SegmentFileStore.FolderDetails();
      folderDetails.setFiles(indexFileNames.get(partition));
      folderDetails.setPartitions(
          Collections.singletonList(partitionLoc.substring(partitionLoc.indexOf("/") + 1)));
      folderDetails.setRelative(isRelativePath);
      folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
      segmentFile.getLocationMap().put(partitionLoc, folderDetails);
      if (finalSegmentFile != null) {
        finalSegmentFile = finalSegmentFile.merge(segmentFile);
      } else {
        finalSegmentFile = segmentFile;
      }
    }
    Objects.requireNonNull(finalSegmentFile);
    String segmentFilesLocation = CarbonTablePath.getSegmentFilesLocation(tablePath);
    CarbonFile locationFile = FileFactory.getCarbonFile(segmentFilesLocation);
    if (!locationFile.exists()) {
      locationFile.mkdirs();
    }
    String segmentFileName = SegmentFileStore.genSegmentFileName(segmentId, timeStamp);
    SegmentFileStore.writeSegmentFile(finalSegmentFile,
        segmentFilesLocation + "/" + segmentFileName + CarbonTablePath.SEGMENT_EXT);
    SegmentFileStore
        .moveFromTempFolder(finalSegmentFile, segmentId + "_" + timeStamp + ".tmp", tablePath);
  }

  /**
   * Write segment information to the segment folder with index file name and
   * corresponding partitions.
   */
  public static void writeSegmentFileForPartitionTable(String tablePath, final String taskNo,
      String location, String timeStamp, List<String> partitionNames, boolean isMergeIndexFlow)
      throws IOException {
    String tempFolderLoc = timeStamp + ".tmp";
    String writePath = CarbonTablePath.getSegmentFilesLocation(tablePath) + "/" + tempFolderLoc;
    CarbonFile carbonFile = FileFactory.getCarbonFile(writePath);
    if (!carbonFile.exists()) {
      carbonFile.mkdirs();
    }
    CarbonFile tempFolder;
    if (isMergeIndexFlow) {
      // To get the index files from the partition path directly.
      // For old stores after merge index is triggered, we wont have '.tmp' folder.
      tempFolder = FileFactory.getCarbonFile(location);
    } else {
      // When mergeindex property is disabled during compaction, we write temporary segment files
      // with index files present in '.tmp' folder of partition path.
      tempFolder = FileFactory
          .getCarbonFile(location + CarbonCommonConstants.FILE_SEPARATOR + tempFolderLoc);
    }

    if ((tempFolder.exists() && partitionNames.size() > 0) || (isMergeIndexFlow
        && partitionNames.size() > 0)) {
      CarbonFile[] carbonFiles = tempFolder.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          return file.getName().startsWith(taskNo) && (
              file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT) || file.getName()
                  .endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT));
        }
      });
      if (carbonFiles != null && carbonFiles.length > 0) {
        boolean isRelative = false;
        if (location.startsWith(tablePath)) {
          location = location.substring(tablePath.length());
          isRelative = true;
        }
        SegmentFile segmentFile = new SegmentFile();
        FolderDetails folderDetails = new FolderDetails();
        folderDetails.setRelative(isRelative);
        folderDetails.setPartitions(partitionNames);
        folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
        setIndexFileNamesToFolderDetails(folderDetails, carbonFiles);
        segmentFile.addPath(location, folderDetails);
        String path = null;
        if (isMergeIndexFlow) {
          // in case of merge index flow, tasks are launched per partition and all the tasks
          // will be written to the same tmp folder, in that case taskNo is not unique.
          // To generate a unique fileName UUID is used
          path = writePath + "/" + CarbonUtil.generateUUID() + CarbonTablePath.SEGMENT_EXT;
        } else {
          path = writePath + "/" + taskNo + CarbonTablePath.SEGMENT_EXT;
        }
        // write segment info to new file.
        writeSegmentFile(segmentFile, path);
      }
    }
  }

  /**
   * Generate Segment file name
   * @param segmentId segment id
   * @param UUID unique string, typically caller can use the loading start
   *             timestamp in CarbonLoadModel
   * @return
   */
  public static String genSegmentFileName(String segmentId, String UUID) {
    return segmentId + "_" + UUID;
  }

  public static String writeSegmentFile(CarbonTable carbonTable, String segmentId, String uuid,
      SegmentMetaDataInfo segmentMetaDataInfo) throws IOException {
    return writeSegmentFile(carbonTable, segmentId, uuid, null, segmentMetaDataInfo);
  }

  public static String writeSegmentFile(CarbonTable carbonTable, String segmentId, String uuid)
      throws IOException {
    return writeSegmentFile(carbonTable, segmentId, uuid, null, null);
  }

  public static String writeSegmentFile(CarbonTable carbonTable, String segmentId, String uuid,
      String segPath) throws IOException {
    return writeSegmentFile(carbonTable, segmentId, uuid, segPath, null);
  }

  /**
   * Returns the list of index files
   *
   * @param segmentPath
   * @return
   */
  public static CarbonFile[] getListOfCarbonIndexFiles(String segmentPath) {
    CarbonFile segmentFolder = FileFactory.getCarbonFile(segmentPath);
    return segmentFolder.listFiles(new CarbonFileFilter() {
      @Override
      public boolean accept(CarbonFile file) {
        return (file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT) ||
            file.getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT));
      }
    });
  }

  /**
   * Write segment file to the metadata folder of the table.
   *
   * @param carbonTable CarbonTable
   * @param segment segment
   * @return boolean , whether write is success or fail.
   */
  public static boolean writeSegmentFile(CarbonTable carbonTable, Segment segment)
      throws IOException {
    String tablePath = carbonTable.getTablePath();
    CarbonFile[] indexFiles = getListOfCarbonIndexFiles(segment.getSegmentPath());
    if (indexFiles != null && indexFiles.length > 0) {
      SegmentFile segmentFile = new SegmentFile();
      segmentFile.setOptions(segment.getOptions());
      FolderDetails folderDetails = new FolderDetails();
      folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
      folderDetails.setRelative(false);
      segmentFile.addPath(segment.getSegmentPath(), folderDetails);
      setIndexFileNamesToFolderDetails(folderDetails, indexFiles);
      String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
      CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
      if (!carbonFile.exists()) {
        carbonFile.mkdirs();
      }
      // write segment info to new file.
      writeSegmentFile(segmentFile,
          segmentFileFolder + File.separator + segment.getSegmentFileName());

      return true;
    }
    return false;
  }

  public static void setIndexFileNamesToFolderDetails(FolderDetails folderDetails,
      CarbonFile[] indexFiles) {
    for (CarbonFile file : indexFiles) {
      if (file.getName().endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
        folderDetails.setMergeFileName(file.getName());
      } else {
        folderDetails.getFiles().add(file.getName());
      }
    }
  }

  public static boolean writeSegmentFileForOthers(
      CarbonTable carbonTable,
      Segment segment,
      PartitionSpec partitionSpec,
      List<FileStatus> partitionDataFiles)
      throws IOException {
    String tablePath = carbonTable.getTablePath();
    CarbonFile[] dataFiles = null;
    if (partitionDataFiles.isEmpty()) {
      CarbonFile segmentFolder = FileFactory.getCarbonFile(segment.getSegmentPath());
      dataFiles = segmentFolder.listFiles(
          file -> (!file.getName().equals("_SUCCESS") && !file.getName().endsWith(".crc")));
    } else {
      dataFiles = partitionDataFiles.stream().map(
          fileStatus -> FileFactory.getCarbonFile(
              fileStatus.getPath().toString())).toArray(CarbonFile[]::new);
    }
    if (dataFiles != null && dataFiles.length > 0) {
      SegmentFile segmentFile = new SegmentFile();
      segmentFile.setOptions(segment.getOptions());
      FolderDetails folderDetails = new FolderDetails();
      folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
      folderDetails.setRelative(false);
      if (!partitionDataFiles.isEmpty()) {
        folderDetails.setPartitions(partitionSpec.getPartitions());
        segmentFile.addPath(partitionSpec.getLocation().toString(), folderDetails);
      } else {
        segmentFile.addPath(segment.getSegmentPath(), folderDetails);
      }
      for (CarbonFile file : dataFiles) {
        folderDetails.getFiles().add(file.getName());
      }
      String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
      CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
      if (!carbonFile.exists()) {
        carbonFile.mkdirs();
      }
      // write segment info to new file.
      writeSegmentFile(segmentFile,
          segmentFileFolder + File.separator + segment.getSegmentFileName());

      return true;
    }
    return false;
  }

  /**
   * Write segment file to the metadata folder of the table selecting only the current load files
   *
   * @param carbonTable
   * @param segmentId
   * @param uuid
   * @return
   * @throws IOException
   */
  public static String writeSegmentFile(CarbonTable carbonTable, String segmentId, String uuid,
        String absSegPath, SegmentMetaDataInfo segmentMetaDataInfo)
      throws IOException {
    String tablePath = carbonTable.getTablePath();
    boolean supportFlatFolder = carbonTable.isSupportFlatFolder();
    String segmentPath = absSegPath;
    if (absSegPath == null) {
      segmentPath = CarbonTablePath.getSegmentPath(tablePath, segmentId);
    }
    CarbonFile segmentFolder = FileFactory.getCarbonFile(segmentPath);
    CarbonFile[] indexFiles = segmentFolder.listFiles(new CarbonFileFilter() {
      @Override
      public boolean accept(CarbonFile file) {
        if (null != uuid) {
          return file.getName().contains(uuid) && (
              file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT) || file.getName()
                  .endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT));
        }
        return file.getName().endsWith(CarbonTablePath.INDEX_FILE_EXT) || file.getName()
            .endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT);
      }
    });
    if (indexFiles != null && indexFiles.length > 0) {
      SegmentFile segmentFile = new SegmentFile();
      FolderDetails folderDetails = new FolderDetails();
      folderDetails.setRelative(absSegPath == null);
      folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
      setIndexFileNamesToFolderDetails(folderDetails, indexFiles);
      String segmentRelativePath = "/";
      if (!supportFlatFolder) {
        if (absSegPath != null) {
          segmentRelativePath = absSegPath;
        } else {
          segmentRelativePath = segmentPath.substring(tablePath.length());
        }
      }
      segmentFile.addPath(segmentRelativePath, folderDetails);
      // set segmentMinMax to segmentFile
      if (null != segmentMetaDataInfo) {
        segmentFile.setSegmentMetaDataInfo(segmentMetaDataInfo);
      }
      String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
      CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
      if (!carbonFile.exists()) {
        carbonFile.mkdirs();
      }
      String segmentFileName = genSegmentFileName(segmentId, uuid) + CarbonTablePath.SEGMENT_EXT;
      // write segment info to new file.
      writeSegmentFile(segmentFile, segmentFileFolder + File.separator + segmentFileName);

      // Move all files to table path from segment folder.
      if (supportFlatFolder) {
        moveFromTempFolder(segmentPath, tablePath);
      }

      return segmentFileName;
    }
    return null;
  }

  /**
   * Get stale and invalid index files that have already been merged.
   * As we are not deleting index files immediately after updating old tables,
   * we will have old index files in segment folder.
   * This method is called in following scenarios:
   * 1. During read, when segment file is not present or gets deleted.
   * 2. When writing segment index size in tablestatus file.
   */
  public static Set<String> getInvalidAndMergedIndexFiles(List<String> indexFiles)
      throws IOException {
    SegmentIndexFileStore indexFileStore = new SegmentIndexFileStore();
    Set<String> mergedAndInvalidIndexFiles = new HashSet<>();
    long lastModifiedTime = 0L;
    String validMergeIndexFile = null;
    List<String> mergeIndexFileNames = new ArrayList<>();
    boolean isIndexFilesPresent = false;
    for (String indexFile : indexFiles) {
      if (indexFile.endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
        isIndexFilesPresent = true;
      }
      if (indexFile.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
        long indexFileTimeStamp =
            Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(indexFile));
        // In case there are more than 1 mergeindex files present, latest one is considered as valid
        if (indexFileTimeStamp > lastModifiedTime) {
          lastModifiedTime = indexFileTimeStamp;
          validMergeIndexFile = indexFile;
        }
        mergeIndexFileNames.add(indexFile);
      }
    }
    //  Possible scenarios where we have >1 merge index file:
    //  1. SI load when MT has stale index files. As during SI load MT segment file is not updated,
    //  we directly read from segment directory.
    //  2. In SI table, after small files index merge(refresh command) we will have more than one
    //  mergeindex file as we are not deleting old file immediately. If segment file gets deleted,
    //  then it reads from segment directory.
    if (mergeIndexFileNames.size() > 1 && validMergeIndexFile != null) {
      final String validIndexFileName = validMergeIndexFile;
      // get the invalid mergeindex files by excluding the valid file.
      mergedAndInvalidIndexFiles.addAll(
          mergeIndexFileNames.stream().filter(file -> !file.equalsIgnoreCase(validIndexFileName))
              .collect(Collectors.toSet()));
    }
    // If both index and merge index files are present, read the valid merge index file and add its
    // mapped index files to the excluding list.
    // Example: We have xxx.index and xxx.mergeindex files in a segment directory.
    // Here the index file (xxx.index) is considered invalid.
    if (isIndexFilesPresent && validMergeIndexFile != null) {
      indexFileStore.readMergeFile(validMergeIndexFile);
      Map<String, List<String>> carbonMergeFileToIndexFilesMap =
          indexFileStore.getCarbonMergeFileToIndexFilesMap();
      String segmentPath =
          validMergeIndexFile.substring(0, validMergeIndexFile.lastIndexOf(File.separator) + 1);
      mergedAndInvalidIndexFiles
          .addAll(carbonMergeFileToIndexFilesMap.get(validMergeIndexFile).stream()
          .map(file -> segmentPath + file).collect(Collectors.toSet()));
    }
    if (mergeIndexFileNames.size() == 0 && indexFiles.size() > 1) {
      // if more than one index file present with different timestamps, then stale/invalid
      // data is present. Latest one is considered as valid.
      Long validFile = indexFiles.stream()
          .map(file -> Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(file)))
          .max(Long::compareTo).get();
      mergedAndInvalidIndexFiles.addAll(
          indexFiles.stream().filter(file -> !file.contains(validFile.toString()))
              .collect(Collectors.toSet()));
    }
    return mergedAndInvalidIndexFiles;
  }

  /**
   * Get valid index files excluding the invalid ones.
   * This method is called while calculating index size using segment directory.
   */
  public static Object getValidCarbonIndexFiles(Object carbonFiles) throws IOException {
    // For supporting local or default file system.
    if (carbonFiles instanceof CarbonFile[]) {
      Set<String> mergedAndInvalidIndexFiles = getInvalidAndMergedIndexFiles(
          Arrays.stream((CarbonFile[]) carbonFiles).map(file -> file.getAbsolutePath())
              .collect(Collectors.toList()));
      if (!mergedAndInvalidIndexFiles.isEmpty()) {
        return Arrays.stream((CarbonFile[]) carbonFiles)
            .filter(file -> !mergedAndInvalidIndexFiles.contains(file.getAbsolutePath()))
            .toArray(CarbonFile[]::new);
      }
    }
    // For supporting HDFS/S3/other file systems.
    else if (carbonFiles instanceof FileStatus[]) {
      Set<String> mergedAndInvalidIndexFiles = getInvalidAndMergedIndexFiles(
          Arrays.stream((FileStatus[]) carbonFiles).map(file -> file.getPath().toString())
              .collect(Collectors.toList()));
      if (!mergedAndInvalidIndexFiles.isEmpty()) {
        return Arrays.stream((FileStatus[]) carbonFiles)
            .filter(file -> !mergedAndInvalidIndexFiles.contains(file.getPath().toString()))
            .toArray(FileStatus[]::new);
      }
    }
    return carbonFiles;
  }

  /**
   * Move the loaded data from source folder to destination folder.
   */
  private static void moveFromTempFolder(String source, String dest) {

    CarbonFile oldFolder = FileFactory.getCarbonFile(source);
    CarbonFile[] oldFiles = oldFolder.listFiles();
    for (CarbonFile file : oldFiles) {
      file.renameForce(dest + CarbonCommonConstants.FILE_SEPARATOR + file.getName());
    }
    oldFolder.delete();
  }

  /**
   * Writes the segment file in json format
   * @param segmentFile
   * @param path
   * @throws IOException
   */
  public static void writeSegmentFile(SegmentFile segmentFile, String path) throws IOException {
    AtomicFileOperations fileWrite =
        AtomicFileOperationFactory.getAtomicFileOperations(path);
    BufferedWriter brWriter = null;
    DataOutputStream dataOutputStream = null;
    Gson gsonObjectToWrite = new Gson();
    try {
      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));

      String metadataInstance = gsonObjectToWrite.toJson(segmentFile);
      brWriter.write(metadataInstance);
      brWriter.flush();
    } catch (IOException ie) {
      LOGGER.error("Error message: " + ie.getLocalizedMessage());
      fileWrite.setFailed();
      throw ie;
    } finally {
      CarbonUtil.closeStreams(brWriter);
      fileWrite.close();
    }
  }

  /**
   * Merge all segment files in a segment to single file.
   *
   * @throws IOException
   */
  public static SegmentFile mergeSegmentFiles(String readPath, String mergeFileName,
      String writePath)
      throws IOException {
    CarbonFile[] segmentFiles = getSegmentFiles(readPath);
    if (segmentFiles != null && segmentFiles.length > 0) {
      SegmentFile segmentFile = mergeSegmentFiles(mergeFileName, writePath, segmentFiles);
      FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(readPath));
      return segmentFile;
    }
    return null;
  }

  public static SegmentFile mergeSegmentFiles(String mergeFileName, String writePath,
      CarbonFile[] segmentFiles) throws IOException {
    SegmentFile segmentFile = null;
    for (CarbonFile file : segmentFiles) {
      SegmentFile localSegmentFile = readSegmentFile(file.getAbsolutePath());
      if (segmentFile == null && localSegmentFile != null) {
        segmentFile = localSegmentFile;
      }
      if (localSegmentFile != null) {
        segmentFile = segmentFile.merge(localSegmentFile);
      }
    }
    if (segmentFile != null) {
      String path = writePath + "/" + mergeFileName + CarbonTablePath.SEGMENT_EXT;
      writeSegmentFile(segmentFile, path);
    }
    return segmentFile;
  }

  public static String getSegmentFilePath(String tablePath, String segmentFileName) {
    return CarbonTablePath.getSegmentFilesLocation(tablePath) +
        CarbonCommonConstants.FILE_SEPARATOR + segmentFileName;
  }

  /**
   * This API will update the segmentFile of a passed segment.
   *
   * @return boolean which determines whether status update is done or not.
   * @throws IOException
   */
  public static boolean updateTableStatusFile(CarbonTable carbonTable, String segmentId,
      String segmentFile, String tableId, SegmentFileStore segmentFileStore,
      String tblStatusReadVersion) throws IOException {
    return updateTableStatusFile(carbonTable, segmentId, segmentFile, tableId, segmentFileStore,
        null, tblStatusReadVersion);
  }

  /**
   * This API will update the table status file with specified segment.
   *
   * @return boolean which determines whether status update is done or not.
   * @throws IOException
   */
  public static boolean updateTableStatusFile(CarbonTable carbonTable, String segmentId,
      String segmentFile, String tableId, SegmentFileStore segmentFileStore,
      SegmentStatus segmentStatus, String tblStatusReadVersion) throws IOException {
    boolean status = false;
    String tablePath = carbonTable.getTablePath();
    String tableStatusPath =
        CarbonTablePath.getTableStatusFilePath(tablePath, tblStatusReadVersion);
    if (!FileFactory.isFileExist(tableStatusPath)) {
      return status;
    }
    String metadataPath = CarbonTablePath.getMetadataPath(tablePath);
    AbsoluteTableIdentifier absoluteTableIdentifier =
        AbsoluteTableIdentifier.from(tablePath, null, null, tableId);
    SegmentStatusManager segmentStatusManager =
        new SegmentStatusManager(absoluteTableIdentifier, carbonTable.getTableStatusVersion());
    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
    int retryCount = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
            CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT);
    int maxTimeout = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
            CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT);
    try {
      if (carbonLock.lockWithRetries(retryCount, maxTimeout)) {
        LOGGER.info("Acquired lock for table path" + tablePath + " for table status update");
        LoadMetadataDetails[] listOfLoadFolderDetailsArray =
            SegmentStatusManager.readLoadMetadata(metadataPath, tblStatusReadVersion);

        for (LoadMetadataDetails detail : listOfLoadFolderDetailsArray) {
          // if the segments is in the list of marked for delete then update the status.
          if (segmentId.equals(detail.getLoadName())) {
            detail.setLoadEndTime(System.currentTimeMillis());
            detail.setSegmentFile(segmentFile);
            if (segmentStatus != null) {
              HashMap<String, Long> dataSizeAndIndexSize =
                  CarbonUtil.getDataSizeAndIndexSize(segmentFileStore, detail.isCarbonFormat());
              detail.setDataSize(
                  dataSizeAndIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_DATA_SIZE)
                      .toString());
              detail.setIndexSize(
                  dataSizeAndIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_INDEX_SIZE)
                      .toString());
              detail.setSegmentStatus(segmentStatus);
            } else {
              detail.setIndexSize(String.valueOf(CarbonUtil
                  .getCarbonIndexSize(segmentFileStore, segmentFileStore.getLocationMap())));
            }
            break;
          }
        }

        SegmentStatusManager
            .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray);
        // clear index cache for the segmentId for which the table status file is getting updated
        clearBlockIndexCache(carbonTable, segmentId);
        status = true;
      } else {
        LOGGER.error(
            "Not able to acquire the lock for Table status update for table path " + tablePath);
      }

    } finally {
      if (carbonLock.unlock()) {
        LOGGER.info("Table unlocked successfully after table status update" + tablePath);
      } else {
        LOGGER.error(
            "Unable to unlock Table lock for table" + tablePath + " during table status update");
      }
    }
    return status;
  }

  /**
   * After updating table status file clear the index cache for all segmentId's on which
   * index is being created because flows like merge index file creation involves modification of
   * segment file and once segment file is modified the cache for that segment need to be cleared
   * otherwise the old cache will be used which is stale
   *
   * @param carbonTable
   * @param segmentId
   */
  public static void clearBlockIndexCache(CarbonTable carbonTable, String segmentId) {
    TableIndex defaultIndex = IndexStoreManager.getInstance().getDefaultIndex(carbonTable);
    LOGGER.info(
        "clearing cache while updating segment file entry in table status file for segmentId: "
            + segmentId);
    defaultIndex.getIndexFactory().clear(segmentId);
  }

  private static CarbonFile[] getSegmentFiles(String segmentPath) {
    CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
    if (carbonFile.exists()) {
      return carbonFile.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          return file.getName().endsWith(CarbonTablePath.SEGMENT_EXT);
        }
      });
    }
    return null;
  }

  /**
   * It provides segment file only for the partitions which has physical index files.
   *
   * @param partitionSpecs
   */
  public static SegmentFile getSegmentFileForPhysicalDataPartitions(String tablePath,
      List<PartitionSpec> partitionSpecs) throws IOException {
    SegmentFile segmentFile = null;
    for (PartitionSpec spec : partitionSpecs) {
      String location = spec.getLocation().toString();
      CarbonFile carbonFile = FileFactory.getCarbonFile(location);

      CarbonFile[] listFiles =
          carbonFile.listFiles(file -> CarbonTablePath.isCarbonIndexFile(file.getAbsolutePath()));
      if (listFiles != null && listFiles.length > 0) {
        Set<String> mergedAndInvalidIndexFiles = getInvalidAndMergedIndexFiles(
            Arrays.stream(listFiles).map(CarbonFile::getAbsolutePath).collect(Collectors.toList()));
        // In add hive partition after merge index is written, delete index files that are merged.
        for (CarbonFile indexFile : listFiles) {
          if (mergedAndInvalidIndexFiles.contains(indexFile.getAbsolutePath())) {
            indexFile.delete();
          }
        }
        CarbonFile[] carbonIndexFiles = listFiles;
        if (!mergedAndInvalidIndexFiles.isEmpty()) {
          carbonIndexFiles = Arrays.stream(listFiles)
              .filter(file -> !mergedAndInvalidIndexFiles.contains(file.getAbsolutePath()))
              .toArray(CarbonFile[]::new);
        }
        boolean isRelative = false;
        if (location.startsWith(tablePath)) {
          location = location.substring(tablePath.length());
          isRelative = true;
        }
        SegmentFile localSegmentFile = new SegmentFile();
        FolderDetails folderDetails = new FolderDetails();
        folderDetails.setRelative(isRelative);
        folderDetails.setPartitions(spec.getPartitions());
        folderDetails.setStatus(SegmentStatus.SUCCESS.getMessage());
        setIndexFileNamesToFolderDetails(folderDetails, carbonIndexFiles);
        localSegmentFile.addPath(location, folderDetails);
        if (segmentFile == null) {
          segmentFile = localSegmentFile;
        } else {
          segmentFile = segmentFile.merge(localSegmentFile);
        }
      }
    }
    return segmentFile;
  }

  /**
   * This method reads the segment file which is written in json format
   *
   * @param segmentFilePath
   * @return
   */
  public static SegmentFile readSegmentFile(String segmentFilePath) throws IOException {
    Gson gsonObjectToRead = new Gson();
    DataInputStream dataInputStream = null;
    BufferedReader buffReader = null;
    InputStreamReader inStream = null;
    SegmentFile segmentFile;
    AtomicFileOperations fileOperation =
        AtomicFileOperationFactory.getAtomicFileOperations(segmentFilePath);

    try {
      if (!FileFactory.isFileExist(segmentFilePath)) {
        return null;
      }
      dataInputStream = fileOperation.openForRead();
      inStream = new InputStreamReader(dataInputStream,
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
      buffReader = new BufferedReader(inStream);
      segmentFile = gsonObjectToRead.fromJson(buffReader, SegmentFile.class);
    } finally {
      if (inStream != null) {
        CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
      }
    }

    return segmentFile;
  }

  /**
   * Reads segment file.
   */
  private SegmentFile readSegment(String tablePath, String segmentFileName) throws IOException {
    String segmentFilePath =
        CarbonTablePath.getSegmentFilesLocation(tablePath) + CarbonCommonConstants.FILE_SEPARATOR
            + segmentFileName;
    return readSegmentFile(segmentFilePath);
  }

  public String getTablePath() {
    return tablePath;
  }

  /**
   * Gets all the index files and related carbondata files from this segment. First user needs to
   * call @readIndexFiles method before calling it.
   * @return
   */
  public Map<String, List<String>> getIndexFilesMap() {
    return indexFilesMap;
  }

  /**
   * Reads all index files which are located in this segment. First user needs to call
   * @readSegment method before calling it.
   * @throws IOException
   */
  public void readIndexFiles(Configuration configuration) throws IOException {
    readIndexFiles(SegmentStatus.SUCCESS, false, configuration);
  }

  public SegmentFile getSegmentFile() {
    return segmentFile;
  }

  /**
   * Reads all index files as per the status of the file. In case of @ignoreStatus is true it just
   * reads all index files
   * @param status
   * @param ignoreStatus
   * @throws IOException
   */
  public List<String> readIndexFiles(SegmentStatus status, boolean ignoreStatus,
      Configuration configuration) throws IOException {
    if (indexFilesMap != null) {
      return new ArrayList<>();
    }
    List<String> indexOrMergeFiles = new ArrayList<>();
    SegmentIndexFileStore indexFileStore = new SegmentIndexFileStore();
    indexFilesMap = new HashMap<>();
    indexFileStore.readAllIIndexOfSegment(this.segmentFile, tablePath, status, ignoreStatus);
    Map<String, byte[]> carbonIndexMap = indexFileStore.getCarbonIndexMapWithFullPath();
    DataFileFooterConverter fileFooterConverter = new DataFileFooterConverter(configuration);
    for (Map.Entry<String, byte[]> entry : carbonIndexMap.entrySet()) {
      List<DataFileFooter> indexInfo =
          fileFooterConverter.getIndexInfo(entry.getKey(), entry.getValue());
      // carbon index file stores blocklets so block filename will be duplicated, use set to remove
      // duplicates
      Set<String> blocks = new LinkedHashSet<>();
      for (DataFileFooter footer : indexInfo) {
        blocks.add(footer.getBlockInfo().getFilePath());
      }
      indexFilesMap.put(entry.getKey(), new ArrayList<>(blocks));
      boolean added = false;
      for (Map.Entry<String, List<String>> mergeFile : indexFileStore
          .getCarbonMergeFileToIndexFilesMap().entrySet()) {
        if (mergeFile.getValue().contains(entry.getKey()
            .substring(entry.getKey().lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR) + 1))) {
          indexOrMergeFiles.add(mergeFile.getKey());
          added = true;
          break;
        }
      }
      if (!added) {
        indexOrMergeFiles.add(entry.getKey());
      }
    }
    return indexOrMergeFiles;
  }

  /**
   * Reads all index files and get the schema of each index file
   * @throws IOException
   */
  public static Map<String, List<ColumnSchema>> getSchemaFiles(SegmentFile segmentFile,
      String tablePath) throws IOException {
    Map<String, List<ColumnSchema>> schemaMap = new HashMap<>();
    if (segmentFile == null) {
      return schemaMap;
    }
    SegmentIndexFileStore indexFileStore = new SegmentIndexFileStore();
    indexFileStore.readAllIIndexOfSegment(segmentFile, tablePath, SegmentStatus.SUCCESS, true);
    Map<String, byte[]> carbonIndexMap = indexFileStore.getCarbonIndexMapWithFullPath();
    DataFileFooterConverter fileFooterConverter = new DataFileFooterConverter();
    for (Map.Entry<String, byte[]> entry : carbonIndexMap.entrySet()) {
      List<DataFileFooter> indexInfo = fileFooterConverter
          .getIndexInfo(entry.getKey(), entry.getValue());
      if (indexInfo.size() > 0) {
        schemaMap.put(entry.getKey(), indexInfo.get(0).getColumnInTable());
      }
    }
    return schemaMap;
  }

  /**
   * Gets all index files from this segment
   * @return
   */
  public Set<String> getIndexFiles() {
    Set<String> indexFiles = new HashSet<>();
    if (segmentFile != null) {
      for (Map.Entry<String, FolderDetails> entry : getLocationMap().entrySet()) {
        String location = entry.getKey();
        if (entry.getValue().isRelative) {
          location = tablePath + location;
        }
        if (entry.getValue().status.equals(SegmentStatus.SUCCESS.getMessage())) {
          for (String indexFile : entry.getValue().getFiles()) {
            indexFiles.add(location + CarbonCommonConstants.FILE_SEPARATOR + indexFile);
          }
        }
      }
    }
    return indexFiles;
  }

  /**
   * Gets all index files from this segment
   * @return
   */
  public Map<String, String> getIndexAndMergeFiles() throws IOException {
    Map<String, String> indexFiles = new HashMap<>();
    if (segmentFile != null) {
      for (Map.Entry<String, FolderDetails> entry : getLocationMap().entrySet()) {
        String location = entry.getKey();
        if (entry.getValue().isRelative) {
          if (location.equals("/")) {
            // in case of flat folder, the relative segment location is '/',
            // so don't append it as we again add file separator for file names.
            location = tablePath;
          } else {
            location = tablePath + location;
          }
        }
        if (entry.getValue().status.equals(SegmentStatus.SUCCESS.getMessage())) {
          String mergeFileName = entry.getValue().getMergeFileName();
          if (null != mergeFileName) {
            indexFiles.put(location + CarbonCommonConstants.FILE_SEPARATOR + mergeFileName,
                entry.getValue().mergeFileName);
          }
          Set<String> files = entry.getValue().getFiles();
          if (null != files && !files.isEmpty()) {
            for (String indexFile : files) {
              String indexFilePath = location + CarbonCommonConstants.FILE_SEPARATOR + indexFile;
              // In the 1.3 store, files field contain the carbonindex files names
              // even if they are merged to a carbonindexmerge file. In that case we have to check
              // for the physical existence of the file to decide
              // on whether it is already merged or not.
              if (FileFactory.isFileExist(indexFilePath)) {
                indexFiles.put(indexFilePath, null);
              }
            }
          }
        }
      }
    }
    return indexFiles;
  }

  /**
   * Gets all carbon index files from this segment
   * @return
   */
  public List<CarbonFile> getIndexCarbonFiles() {
    Set<String> indexFiles = getIndexFiles();
    List<CarbonFile> carbonFiles = new ArrayList<>();
    for (String indexFile : indexFiles) {
      CarbonFile carbonFile = FileFactory.getCarbonFile(indexFile);
      if (carbonFile.exists()) {
        carbonFiles.add(carbonFile);
      }
    }
    return carbonFiles;
  }

  /**
   * Drops the partition related files from the segment file of the segment and writes
   * to a new file. First iterator over segment file and check the path it needs to be dropped.
   * And update the status with delete if it found.
   *
   * @param uniqueId
   * @throws IOException
   */
  public void dropPartitions(String segmentNo, List<String> partitionLocations,
      String uniqueId, List<String> toBeDeletedSegments, List<String> toBeUpdatedSegments)
      throws IOException {
    boolean updateSegment = false;
    for (Map.Entry<String, FolderDetails> entry : segmentFile.getLocationMap().entrySet()) {
      String location = entry.getKey();
      if (entry.getValue().isRelative) {
        location = tablePath + CarbonCommonConstants.FILE_SEPARATOR + location;
      }
      Path path = new Path(location);
      // Update the status to delete if path equals
      if (null != partitionLocations) {
        for (String partitionLocation : partitionLocations) {
          if (path.toString().equals(partitionLocation)) {
            entry.getValue().setStatus(SegmentStatus.MARKED_FOR_DELETE.getMessage());
            updateSegment = true;
            break;
          }
        }
      }
    }
    if (updateSegment) {
      String writePath = CarbonTablePath.getSegmentFilesLocation(tablePath);
      writePath =
          writePath + CarbonCommonConstants.FILE_SEPARATOR +
              SegmentFileStore.genSegmentFileName(segmentNo,  String.valueOf(uniqueId))
              + CarbonTablePath.SEGMENT_EXT;
      writeSegmentFile(segmentFile, writePath);
    }
    // Check whether we can completely remove the segment.
    boolean deleteSegment = true;
    for (Map.Entry<String, FolderDetails> entry : segmentFile.getLocationMap().entrySet()) {
      if (entry.getValue().getStatus().equals(SegmentStatus.SUCCESS.getMessage())) {
        deleteSegment = false;
        break;
      }
    }
    if (deleteSegment) {
      toBeDeletedSegments.add(segmentNo);
    }
    if (updateSegment) {
      toBeUpdatedSegments.add(segmentNo);
    }
  }

  /**
   * Update the table status file with the dropped partitions information
   *
   * @param carbonTable
   * @param uniqueId
   * @param toBeUpdatedSegments
   * @param toBeDeleteSegments
   * @throws IOException
   */
  public static String commitDropPartitions(CarbonTable carbonTable, String uniqueId,
      List<String> toBeUpdatedSegments, List<String> toBeDeleteSegments, String uuid)
      throws IOException {
    String tblStatusWriteVersion = "";
    if (toBeDeleteSegments.size() > 0 || toBeUpdatedSegments.size() > 0) {
      Set<Segment> segmentSet = new HashSet<>(
          new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(),
              carbonTable.getTableStatusVersion()).getValidAndInvalidSegments(carbonTable.isMV())
              .getValidSegments());
      Map<String, String> returnTuple =
          CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, uniqueId, true, false,
              Segment.toSegmentList(toBeDeleteSegments, null),
              Segment.toSegmentList(toBeUpdatedSegments, null), uuid);
      tblStatusWriteVersion = returnTuple.getOrDefault("tblStatusWriteVersion", "");
    }
    return tblStatusWriteVersion;
  }

  /**
   * Clean up invalid data after drop partition in partial segments of table
   *
   * @param table the table need to cleaned stale data
   * @param details the segments need to cleaned stale data
   * @param partitionSpecs the partitions need to cleaned stale data
   * @param forceDelete Whether it should be deleted force or check the time for an hour creation
   *                    to delete data.
   * @throws IOException
   */
  public static void cleanSegments(CarbonTable table,
      LoadMetadataDetails[] details,
      List<PartitionSpec> partitionSpecs,
      boolean forceDelete) throws IOException {
    // scan through each segment.
    for (LoadMetadataDetails segment : details) {
      // if this segment is valid then only we will go for deletion of related
      // dropped partition files. 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)
          && segment.getSegmentFile() != null) {
        List<String> toBeDeletedIndexFiles = new ArrayList<>();
        List<String> toBeDeletedDataFiles = new ArrayList<>();
        // take the list of files from this segment.
        SegmentFileStore fileStore =
            new SegmentFileStore(table.getTablePath(), segment.getSegmentFile());
        List<String> indexOrMergeFiles = fileStore
            .readIndexFiles(SegmentStatus.MARKED_FOR_DELETE, false, FileFactory.getConfiguration());
        if (forceDelete) {
          deletePhysicalPartition(
              partitionSpecs,
              fileStore.getIndexFilesMap(),
              indexOrMergeFiles,
              table.getTablePath());
        }
        for (Map.Entry<String, List<String>> entry : fileStore.indexFilesMap.entrySet()) {
          String indexFile = entry.getKey();
          // Check the partition information in the partition mapper
          Long fileTimestamp = CarbonUpdateUtil.getTimeStampAsLong(indexFile
              .substring(indexFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
                  indexFile.length() - CarbonTablePath.INDEX_FILE_EXT.length()));
          if (TrashUtil.isDataOutsideTrashIsExpired(fileTimestamp) || forceDelete) {
            // Add the corresponding carbondata files to the delete list.
            toBeDeletedDataFiles.addAll(entry.getValue());
          }
        }
        for (String indexFile : indexOrMergeFiles) {
          Long fileTimestamp = 0L;
          if (indexFile.endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
            fileTimestamp = CarbonUpdateUtil.getTimeStampAsLong(indexFile
                .substring(indexFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
                    indexFile.length() - CarbonTablePath.INDEX_FILE_EXT.length()));
          } else if (indexFile.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)) {
            fileTimestamp = CarbonUpdateUtil.getTimeStampAsLong(indexFile
                .substring(indexFile.lastIndexOf(CarbonCommonConstants.UNDERSCORE) + 1,
                    indexFile.length() - CarbonTablePath.MERGE_INDEX_FILE_EXT.length()));
          }
          if (TrashUtil.isDataOutsideTrashIsExpired(fileTimestamp) || forceDelete) {
            toBeDeletedIndexFiles.add(indexFile);
          }
        }
        if (toBeDeletedIndexFiles.size() > 0) {
          for (String dataFile : toBeDeletedIndexFiles) {
            FileFactory.deleteFile(dataFile);
          }
          for (String dataFile : toBeDeletedDataFiles) {
            FileFactory.deleteFile(dataFile);
          }
        }
      }
    }
  }

  /**
   * Clean up invalid data after drop partition in all segments of table
   *
   * @param table the table need to clean stale data
   * @param partitionSpecs the partitionSpecs need to clean
   * @param forceDelete Whether it should be deleted force or check the time for an hour creation
   *                    to delete data.
   * @throws IOException
   */
  public static void cleanSegments(CarbonTable table, List<PartitionSpec> partitionSpecs,
      boolean forceDelete) throws IOException {
    LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(table.getMetadataPath(),
        table.getTableStatusVersion());
    cleanSegments(table, details, partitionSpecs, forceDelete);
  }

  /**
   * Clean up invalid data after drop partition in partial segments of table
   *
   * @param table the table need to clean stale data
   * @param segments the segments need to clean stale data
   * @param forceDelete Whether it should be deleted force or check the time for an hour creation
   *                    to delete data.
   * @throws IOException
   */
  public static void cleanSegments(CarbonTable table,
      Set<String> segments,
      List<PartitionSpec> partitionSpecs,
      boolean forceDelete) throws IOException {
    LoadMetadataDetails[] details = Arrays.stream(
            SegmentStatusManager.readLoadMetadata(table.getMetadataPath(),
                table.getTableStatusVersion()))
        .filter(detail -> segments.contains(detail.getLoadName()))
        .toArray(LoadMetadataDetails[]::new);
    cleanSegments(table, details, partitionSpecs, forceDelete);
  }

  public static void deleteSegmentFile(String tablePath, Segment segment) throws Exception {
    String segmentFilePath =
        CarbonTablePath.getSegmentFilePath(tablePath, segment.getSegmentFileName());
    // Deletes the physical segment file
    FileFactory.deleteFile(segmentFilePath);
  }

  /**
   * Deletes the segment file and its physical files like partition folders from disk
   * @param tablePath
   * @param segment
   * @param partitionSpecs
   * @throws IOException
   */
  public static void deleteSegment(String tablePath, Segment segment,
      List<PartitionSpec> partitionSpecs,
      SegmentUpdateStatusManager updateStatusManager) throws Exception {
    SegmentFileStore fileStore = new SegmentFileStore(tablePath, segment.getSegmentFileName());
    List<String> indexOrMergeFiles = fileStore.readIndexFiles(SegmentStatus.SUCCESS, true,
        FileFactory.getConfiguration());
    Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
    List<String> deletedFiles = new ArrayList<>();
    for (String indexFilePath : indexOrMergeFiles) {
      FileFactory.deleteFile(indexFilePath);
      deletedFiles.add(indexFilePath);
    }
    for (Map.Entry<String, List<String>> entry : indexFilesMap.entrySet()) {
      for (String file : entry.getValue()) {
        String[] deltaFilePaths =
            updateStatusManager.getDeleteDeltaFilePath(file, segment.getSegmentNo());
        for (String deltaFilePath : deltaFilePaths) {
          FileFactory.deleteFile(deltaFilePath);
          deletedFiles.add(deltaFilePath);
        }
        FileFactory.deleteFile(file);
        deletedFiles.add(file);
      }
    }
    LOGGER.info("Deleted the files: " + String.join(",", deletedFiles) + " on clean" +
        " files operation");
    deletePhysicalPartition(partitionSpecs, indexFilesMap, indexOrMergeFiles, tablePath);
  }

  /**
   * If partition specs are available, then check the location map for any index file path which is
   * not present in the partitionSpecs. If found then delete that index file.
   * If the partition directory is empty, then delete the directory also.
   * If partition specs are null, then directly delete parent directory in locationMap.
   */
  private static void deletePhysicalPartition(List<PartitionSpec> partitionSpecs,
      Map<String, List<String>> locationMap, List<String> indexOrMergeFiles, String tablePath) {
    LOGGER.info("Deleting files: ");
    for (String indexOrMergeFile : indexOrMergeFiles) {
      if (null != partitionSpecs) {
        Path location = new Path(indexOrMergeFile);
        boolean exists = pathExistsInPartitionSpec(partitionSpecs, location);
        if (!exists) {
          FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(location.toString()));
          LOGGER.info(location.toString());
        }
      } else {
        Path location = new Path(indexOrMergeFile);
        FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(location.toString()));
        LOGGER.info(location.toString());
      }
    }
    for (Map.Entry<String, List<String>> entry : locationMap.entrySet()) {
      if (partitionSpecs != null) {
        Path location = new Path(entry.getKey());
        boolean exists = pathExistsInPartitionSpec(partitionSpecs, location);
        if (!exists) {
          FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(location.toString()));
          for (String carbonDataFile : entry.getValue()) {
            FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(carbonDataFile));
          }
        }
        CarbonFile path = FileFactory.getCarbonFile(location.getParent().toString());
        deleteEmptyPartitionFolders(path);
      } else {
        Path location = new Path(entry.getKey()).getParent();
        // delete the segment folder
        CarbonFile segmentPath = FileFactory.getCarbonFile(location.toString());
        if (null != segmentPath && segmentPath.exists() &&
            !new Path(tablePath).equals(new Path(segmentPath.getAbsolutePath()))) {
          FileFactory.deleteAllCarbonFilesOfDir(segmentPath);
        }
      }
    }
  }

  /**
   * This method deletes the directories recursively if there are no files under corresponding
   * folder.
   * Ex: If partition folder is year=2015, month=2,day=5 and drop partition is day=5, it will delete
   * till year partition folder if there are no other folder or files present under each folder till
   * year partition
   */
  public static void deleteEmptyPartitionFolders(CarbonFile path) {
    if (path != null && path.listFiles().length == 0) {
      FileFactory.deleteAllCarbonFilesOfDir(path);
      Path parentsLocation = new Path(path.getAbsolutePath()).getParent();
      deleteEmptyPartitionFolders(
          FileFactory.getCarbonFile(parentsLocation.toString()));
    }
  }

  public static boolean pathExistsInPartitionSpec(List<PartitionSpec> partitionSpecs,
      Path partitionPath) {
    for (PartitionSpec spec : partitionSpecs) {
      if (spec.getLocation().equals(partitionPath)) {
        return true;
      }
    }
    return false;
  }

  /**
   * Get the partition specs of the segment
   * @param segmentId
   * @param tablePath
   * @return
   * @throws IOException
   */
  public static List<PartitionSpec> getPartitionSpecs(String segmentId, String tablePath,
      String segmentFilePath, String loadStartTime) throws IOException {
    SegmentFileStore fileStore = new SegmentFileStore(tablePath, segmentFilePath);
    List<PartitionSpec> partitionSpecs = fileStore.getPartitionSpecs();
    for (PartitionSpec spec : partitionSpecs) {
      spec.setUuid(segmentId + "_" + loadStartTime);
    }
    return partitionSpecs;
  }

  /**
   * Get the partition specs of the segment
   * @param segmentId
   * @param tablePath
   * @return
   * @throws IOException
   */
  public static List<PartitionSpec> getPartitionSpecs(String segmentId, String tablePath,
      LoadMetadataDetails[] details)
      throws IOException {
    LoadMetadataDetails segEntry = null;
    for (LoadMetadataDetails entry : details) {
      if (entry.getLoadName().equals(segmentId)) {
        segEntry = entry;
        break;
      }
    }
    if (segEntry != null && segEntry.getSegmentFile() != null) {
      SegmentFileStore fileStore = new SegmentFileStore(tablePath, segEntry.getSegmentFile());
      List<PartitionSpec> partitionSpecs = fileStore.getPartitionSpecs();
      for (PartitionSpec spec : partitionSpecs) {
        spec.setUuid(segmentId + "_" + segEntry.getLoadStartTime());
      }
      return partitionSpecs;
    }
    return new ArrayList<>();
  }

  /**
   * Move the loaded data from temp folder to respective partition folder.
   * @param segmentFile
   * @param tmpFolder
   * @param tablePath
   */
  public static void moveFromTempFolder(SegmentFile segmentFile, String tmpFolder,
      String tablePath) {

    for (Map.Entry<String, SegmentFileStore.FolderDetails> entry : segmentFile.getLocationMap()
        .entrySet()) {
      String location = entry.getKey();
      if (entry.getValue().isRelative()) {
        location = tablePath + CarbonCommonConstants.FILE_SEPARATOR + location;
      }
      CarbonFile oldFolder =
          FileFactory.getCarbonFile(location + CarbonCommonConstants.FILE_SEPARATOR + tmpFolder);
      CarbonFile[] oldFiles = oldFolder.listFiles();
      for (CarbonFile file : oldFiles) {
        file.renameForce(location + CarbonCommonConstants.FILE_SEPARATOR + file.getName());
      }
      oldFolder.delete();
    }
  }

  /**
   * Remove temp stage folder in case of job aborted.
   *
   * @param locationMap
   * @param tmpFolder
   * @param tablePath
   */
  public static void removeTempFolder(Map<String, FolderDetails> locationMap, String tmpFolder,
      String tablePath) {
    if (locationMap == null) {
      return;
    }
    for (Map.Entry<String, SegmentFileStore.FolderDetails> entry : locationMap.entrySet()) {
      String location = entry.getKey();
      if (entry.getValue().isRelative()) {
        location = tablePath + CarbonCommonConstants.FILE_SEPARATOR + location;
      }
      CarbonFile oldFolder =
          FileFactory.getCarbonFile(location + CarbonCommonConstants.FILE_SEPARATOR + tmpFolder);
      if (oldFolder.exists()) {
        FileFactory.deleteAllCarbonFilesOfDir(oldFolder);
      }
    }
  }

  /**
   * Returns content of segment
   * @return
   */
  public Map<String, FolderDetails> getLocationMap() {
    if (segmentFile == null) {
      return new HashMap<>();
    }
    return segmentFile.getLocationMap();
  }

  /**
   * get the current partition specs of this segment
   * @return
   */
  public List<PartitionSpec> getPartitionSpecs() {
    List<PartitionSpec> partitionSpecs = new ArrayList<>();
    if (segmentFile != null) {
      for (Map.Entry<String, FolderDetails> entry : segmentFile.getLocationMap().entrySet()) {
        String location = entry.getKey();
        if (entry.getValue().isRelative) {
          location = tablePath + CarbonCommonConstants.FILE_SEPARATOR + location;
        }
        if (entry.getValue().getStatus().equals(SegmentStatus.SUCCESS.getMessage())) {
          partitionSpecs.add(new PartitionSpec(entry.getValue().partitions, location));
        }
      }
    }
    return partitionSpecs;
  }

  /**
   * It contains the segment information like location, partitions and related index files
   */
  public static class SegmentFile implements Serializable {

    private static final long serialVersionUID = 3582245668420401089L;

    /**
     * mapping of index file parent folder to the index file folder info
     */
    private Map<String, FolderDetails> locationMap;

    /**
     * Segment option properties
     */
    private Map<String, String> options;

    /**
     * Segment metadata information such as column_id, min-max, alter properties
     */
    private String segmentMetaDataInfo;

    public SegmentFile() {
      locationMap = new HashMap<>();
    }

    public SegmentFile merge(SegmentFile segmentFile) throws IOException {
      if (this == segmentFile) {
        return this;
      }
      if (locationMap != null && segmentFile.locationMap != null) {
        for (Map.Entry<String, FolderDetails> entry : segmentFile.locationMap.entrySet()) {
          FolderDetails folderDetails = locationMap.get(entry.getKey());
          if (folderDetails != null) {
            folderDetails.merge(entry.getValue());
          } else {
            locationMap.put(entry.getKey(), entry.getValue());
          }
        }
        if (segmentMetaDataInfo != null) {
          SegmentMetaDataInfo currentSegmentMetaDataInfo =
              (SegmentMetaDataInfo) ObjectSerializationUtil
                  .convertStringToObject(segmentMetaDataInfo);
          if (null != segmentFile.getSegmentMetaDataInfo()) {
            // get updated segmentColumnMetaDataInfo based on comparing block min-max values
            Map<String, SegmentColumnMetaDataInfo> previousBlockColumnMetaDataInfo =
                segmentFile.getSegmentMetaDataInfo().getSegmentColumnMetaDataInfoMap();
            for (Map.Entry<String, SegmentColumnMetaDataInfo> entry :
                previousBlockColumnMetaDataInfo.entrySet()) {
              if (currentSegmentMetaDataInfo.getSegmentColumnMetaDataInfoMap()
                  .containsKey(entry.getKey())) {
                SegmentColumnMetaDataInfo currentBlockMinMaxInfo =
                    currentSegmentMetaDataInfo.getSegmentColumnMetaDataInfoMap()
                        .get(entry.getKey());
                byte[] blockMaxValue = SegmentMetaDataInfoStats.getInstance()
                    .compareAndUpdateMinMax(currentBlockMinMaxInfo.getColumnMaxValue(),
                        entry.getValue().getColumnMaxValue(), false);
                byte[] blockMinValue = SegmentMetaDataInfoStats.getInstance()
                    .compareAndUpdateMinMax(currentBlockMinMaxInfo.getColumnMinValue(),
                        entry.getValue().getColumnMinValue(), true);
                currentSegmentMetaDataInfo.getSegmentColumnMetaDataInfoMap().get(entry.getKey())
                    .setColumnMaxValue(blockMaxValue);
                currentSegmentMetaDataInfo.getSegmentColumnMetaDataInfoMap().get(entry.getKey())
                    .setColumnMinValue(blockMinValue);
              }
            }
          }
          segmentMetaDataInfo =
              ObjectSerializationUtil.convertObjectToString(currentSegmentMetaDataInfo);
        }
      }
      if (locationMap == null) {
        locationMap = segmentFile.locationMap;
      }
      return this;
    }

    public Map<String, FolderDetails> getLocationMap() {
      return locationMap;
    }

    /**
     * Add index file parent folder and the index file folder info
     */
    void addPath(String path, FolderDetails details) {
      locationMap.put(path, details);
    }

    public Map<String, String> getOptions() {
      return options;
    }

    public void setOptions(Map<String, String> options) {
      this.options = options;
    }

    public SegmentMetaDataInfo getSegmentMetaDataInfo() {
      SegmentMetaDataInfo newSegmentMetaDataInfo = null;
      try {
        newSegmentMetaDataInfo = (SegmentMetaDataInfo) ObjectSerializationUtil
            .convertStringToObject(segmentMetaDataInfo);
      } catch (IOException e) {
        LOGGER.error("Error while getting segment metadata info");
      }
      return newSegmentMetaDataInfo;
    }

    public void setSegmentMetaDataInfo(SegmentMetaDataInfo segmentMetaDataInfo) throws IOException {
      this.segmentMetaDataInfo = ObjectSerializationUtil.convertObjectToString(
          segmentMetaDataInfo);
    }
  }

  public static SegmentFile createSegmentFile(String partitionPath, FolderDetails folderDetails) {
    SegmentFile segmentFile = new SegmentFile();
    segmentFile.addPath(partitionPath, folderDetails);
    return segmentFile;
  }

  /**
   * Represents one partition folder
   */
  public static class FolderDetails implements Serializable {

    private static final long serialVersionUID = 501021868886928553L;

    /**
     * Based on isRelative variable:
     * 1. if it is relative, it is relative path to the table path, for all index files
     * 2. if it is not relative, it is the full path of all index files
     */
    private Set<String> files = new HashSet<>();

    /**
     * all partition names
     */
    private List<String> partitions = new ArrayList<>();

    /**
     * status for the partition, success or mark for delete
     */
    private String status;

    /**
     * file name for merge index file in this folder
     */
    private String mergeFileName;

    /**
     * true if it is relative path, for example, if user give partition location when
     * adding the partition, it will be false
     */
    private boolean isRelative;

    public FolderDetails merge(FolderDetails folderDetails) {
      if (this == folderDetails || folderDetails == null) {
        return this;
      }
      if (folderDetails.files != null) {
        files.addAll(folderDetails.files);
      }
      if (files == null) {
        files = folderDetails.files;
      }
      partitions = folderDetails.partitions;
      return this;
    }

    public Set<String> getFiles() {
      return files;
    }

    public void setFiles(Set<String> files) {
      this.files = files;
    }

    public List<String> getPartitions() {
      return partitions;
    }

    public void setPartitions(List<String> partitions) {
      this.partitions = partitions;
    }

    public boolean isRelative() {
      return isRelative;
    }

    public void setRelative(boolean relative) {
      isRelative = relative;
    }

    public String getMergeFileName() {
      return mergeFileName;
    }

    public void setMergeFileName(String mergeFileName) {
      this.mergeFileName = mergeFileName;
    }

    public String getStatus() {
      return status;
    }

    public void setStatus(String status) {
      this.status = status;
    }
  }
}