content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.carbondata.processing.util;
  
  import java.io.IOException;
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  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.CarbonLoadOptionConstants;
  import org.apache.carbondata.core.datastore.block.Distributable;
  import org.apache.carbondata.core.datastore.block.TableBlockInfo;
  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.CarbonLockUtil;
  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.CarbonUpdateUtil;
  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.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.path.CarbonTablePath;
  import org.apache.carbondata.core.writer.CarbonIndexFileMergeWriter;
  import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
  import org.apache.carbondata.processing.merger.NodeMultiBlockRelation;
  
  import static org.apache.carbondata.core.enums.EscapeSequences.BACKSPACE;
  import static org.apache.carbondata.core.enums.EscapeSequences.CARRIAGE_RETURN;
  import static org.apache.carbondata.core.enums.EscapeSequences.NEW_LINE;
  import static org.apache.carbondata.core.enums.EscapeSequences.TAB;
  
  import org.apache.commons.lang3.StringUtils;
  import org.apache.log4j.Logger;
  
  public final class CarbonLoaderUtil {
  
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(CarbonLoaderUtil.class.getName());
  
    private CarbonLoaderUtil() {
    }
  
    /**
     * strategy for assign blocks to nodes/executors
     */
    public enum BlockAssignmentStrategy {
      BLOCK_NUM_FIRST("Assign blocks to node base on number of blocks"),
      BLOCK_SIZE_FIRST("Assign blocks to node base on data size of blocks"),
      NODE_MIN_SIZE_FIRST("Assign blocks to node base on minimum size of inputs");
      private String name;
      BlockAssignmentStrategy(String name) {
        this.name = name;
      }
  
      @Override
      public String toString() {
        return this.getClass().getSimpleName() + ':' + this.name;
      }
    }
  
    /**
     * delete segment after data loading failure
     */
    public static void deleteSegmentForFailure(CarbonLoadModel loadModel) {
      int currentLoad = Integer.parseInt(loadModel.getSegmentId());
      String segmentPath = CarbonTablePath.getSegmentPath(
          loadModel.getTablePath(), currentLoad + "");
      try {
        if (FileFactory.isFileExist(segmentPath)) {
          CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
          CarbonUtil.deleteFoldersAndFiles(carbonFile);
        }
      } catch (IOException | InterruptedException e) {
        LOGGER.error("Unable to delete segment: " + segmentPath + ", " + e.getMessage(), e);
      }
    }
  
    /**
     * the method returns true if the segment has carbondata file else returns false.
     *
     * @param loadModel
     * @param currentLoad
     * @return
     */
    public static boolean isValidSegment(CarbonLoadModel loadModel,
        int currentLoad) {
  
      int fileCount = 0;
      String segmentPath = CarbonTablePath.getSegmentPath(
          loadModel.getTablePath(), currentLoad + "");
      CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
      CarbonFile[] files = carbonFile.listFiles(new CarbonFileFilter() {
  
        @Override
        public boolean accept(CarbonFile file) {
          return file.getName().endsWith(
              CarbonTablePath.getCarbonIndexExtension())
              || file.getName().endsWith(
              CarbonTablePath.getCarbonDataExtension());
        }
  
      });
      fileCount += files.length;
      if (files.length > 0) {
        return true;
      }
      if (fileCount == 0) {
        return false;
      }
      return true;
    }
  
    /**
     * This API will write the load level metadata for the loadmanagement module inorder to
     * manage the load and query execution management smoothly.
     *
     * @param newMetaEntry
     * @param loadModel
     * @return boolean which determines whether status update is done or not.
     * @throws IOException
     */
    public static boolean recordNewLoadMetadata(LoadMetadataDetails newMetaEntry,
        CarbonLoadModel loadModel, boolean loadStartEntry, boolean insertOverwrite)
        throws IOException {
      return recordNewLoadMetadata(newMetaEntry, loadModel, loadStartEntry, insertOverwrite, "",
          false);
    }
  
    /**
     * Append a new load metadata into table status file
     *
     * @param loadModel load model
     * @return boolean which determines whether status update is done or not
     * @throws IOException
     */
    public static boolean recordNewLoadMetadata(CarbonLoadModel loadModel) throws IOException {
      LoadMetadataDetails newLoadMetaEntry = new LoadMetadataDetails();
      loadModel.setFactTimeStamp(System.currentTimeMillis());
      CarbonLoaderUtil.populateNewLoadMetaEntry(
          newLoadMetaEntry,
          SegmentStatus.INSERT_IN_PROGRESS,
          loadModel.getFactTimeStamp(),
          false);
      return recordNewLoadMetadata(newLoadMetaEntry, loadModel, true, false);
    }
  
    /**
     * This API deletes the content of the non Transactional Tables when insert overwrite is set true.
     *
     * @param loadModel
     * @throws IOException
     */
    public static void deleteNonTransactionalTableForInsertOverwrite(final CarbonLoadModel loadModel)
        throws IOException {
      // We need to delete the content of the Table Path Folder except the
      // Newly added file.
      List<String> filesToBeDeleted = new ArrayList<>();
      CarbonFile carbonFile = FileFactory.getCarbonFile(loadModel.getTablePath());
      CarbonFile[] filteredList = carbonFile.listFiles(new CarbonFileFilter() {
        @Override
        public boolean accept(CarbonFile file) {
          return !file.getName().contains(loadModel.getFactTimeStamp() + "");
        }
      });
      for (CarbonFile file : filteredList) {
        filesToBeDeleted.add(file.getAbsolutePath());
      }
  
      deleteFiles(filesToBeDeleted);
    }
  
    /**
     * This API will write the load level metadata for the loadmanagement module inorder to
     * manage the load and query execution management smoothly.
     *
     * @param newMetaEntry
     * @param loadModel
     * @param uuid
     * @return boolean which determines whether status update is done or not.
     * @throws IOException
     */
    public static boolean recordNewLoadMetadata(LoadMetadataDetails newMetaEntry,
        final CarbonLoadModel loadModel, boolean loadStartEntry, boolean insertOverwrite, String uuid,
        boolean isUpdateStatusRequired) throws IOException {
      // For Non Transactional tables no need to update the the Table Status file.
      if (!loadModel.isCarbonTransactionalTable()) {
        return true;
      }
  
      return recordNewLoadMetadata(newMetaEntry, loadModel, loadStartEntry, insertOverwrite, uuid,
          new ArrayList<Segment>(), new ArrayList<Segment>(), isUpdateStatusRequired);
    }
  
    /**
     * This API will write the load level metadata for the loadmanagement module inorder to
     * manage the load and query execution management smoothly.
     *
     * @param newMetaEntry
     * @param loadModel
     * @param uuid
     * @return boolean which determines whether status update is done or not.
     * @throws IOException
     */
    public static boolean recordNewLoadMetadata(LoadMetadataDetails newMetaEntry,
        CarbonLoadModel loadModel, boolean loadStartEntry, boolean insertOverwrite, String uuid,
        List<Segment> segmentsToBeDeleted, List<Segment> segmentFilesTobeUpdated,
        boolean isUpdateStatusRequired) throws IOException {
      boolean status = false;
      AbsoluteTableIdentifier identifier =
          loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
      if (loadModel.isCarbonTransactionalTable()) {
        String metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath());
        if (!FileFactory.isFileExist(metadataPath)) {
          FileFactory.mkdirs(metadataPath);
        }
      }
      SegmentStatusManager segmentStatusManager = new SegmentStatusManager(identifier,
          loadModel.getCarbonDataLoadSchema().getCarbonTable().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" + loadModel.getDatabaseName() + "." + loadModel.getTableName()
                  + " for table status updation");
          loadModel.setLatestTableStatusWriteVersion(String.valueOf(System.currentTimeMillis()));
          String newTblStatusVersion = loadModel.getLatestTableStatusWriteVersion();
          String tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath(),
              loadModel.getLatestTableStatusWriteVersion());
          if (newMetaEntry.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS
              || newMetaEntry.getSegmentStatus() == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) {
            newTblStatusVersion =
                loadModel.getCarbonDataLoadSchema().getCarbonTable().getTableStatusVersion();
          }
          LoadMetadataDetails[] listOfLoadFolderDetailsArray =
              SegmentStatusManager.readLoadMetadata(
                  CarbonTablePath.getMetadataPath(identifier.getTablePath()), newTblStatusVersion);
          List<LoadMetadataDetails> listOfLoadFolderDetails =
              new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
          Collections.addAll(listOfLoadFolderDetails, listOfLoadFolderDetailsArray);
          // create a new segment Id if load has just begun else add the already generated Id
          if (loadStartEntry) {
            String segmentId =
                String.valueOf(SegmentStatusManager.createNewSegmentId(listOfLoadFolderDetailsArray));
            loadModel.setLoadMetadataDetails(listOfLoadFolderDetails);
            LoadMetadataDetails entryTobeRemoved = null;
            if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isMV()
                && !loadModel.getSegmentId().isEmpty()) {
              for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
                if (entry.getLoadName().equalsIgnoreCase(loadModel.getSegmentId())) {
                  newMetaEntry.setLoadName(loadModel.getSegmentId());
                  newMetaEntry.setExtraInfo(entry.getExtraInfo());
                  entryTobeRemoved = entry;
                }
              }
            } else {
              if (isUpdateStatusRequired && segmentId.equalsIgnoreCase("0") && !StringUtils
                  .isBlank(uuid)) {
                newMetaEntry.setUpdateStatusFileName(CarbonUpdateUtil.getUpdateStatusFileName(uuid));
              }
              newMetaEntry.setLoadName(segmentId);
              loadModel.setSegmentId(segmentId);
            }
            listOfLoadFolderDetails.remove(entryTobeRemoved);
            // Exception should be thrown if:
            // 1. If insert overwrite is in progress and any other load or insert operation
            // is triggered
            // 2. If load or insert into operation is in progress and insert overwrite operation
            // is triggered
            for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
              if (entry.getSegmentStatus() == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS
                  && SegmentStatusManager.isLoadInProgress(
                      identifier, entry.getLoadName())) {
                throw new RuntimeException("Already insert overwrite is in progress");
              } else if (newMetaEntry.getSegmentStatus() == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS
                  && entry.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS
                  && SegmentStatusManager.isLoadInProgress(
                      identifier, entry.getLoadName())) {
                throw new RuntimeException("Already insert into or load is in progress");
              }
            }
            listOfLoadFolderDetails.add(newMetaEntry);
          } else {
            newMetaEntry.setLoadName(String.valueOf(loadModel.getSegmentId()));
            // existing entry needs to be overwritten as the entry will exist with some
            // intermediate status
            int indexToOverwriteNewMetaEntry = 0;
            boolean found = false;
            for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
              if (entry.getLoadName().equals(newMetaEntry.getLoadName())
                  && entry.getLoadStartTime() == newMetaEntry.getLoadStartTime()) {
                newMetaEntry.setExtraInfo(entry.getExtraInfo());
                found = true;
                break;
              }
              indexToOverwriteNewMetaEntry++;
            }
            if (insertOverwrite) {
              for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
                if (entry.getSegmentStatus() != SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) {
                  entry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
                }
              }
            }
            if (!found) {
              LOGGER.error("Entry not found to update " + newMetaEntry + " From list :: "
                  + listOfLoadFolderDetails);
              throw new IOException("Entry not found to update in the table status file");
            }
            listOfLoadFolderDetails.set(indexToOverwriteNewMetaEntry, newMetaEntry);
          }
  
          for (LoadMetadataDetails detail: listOfLoadFolderDetails) {
            // if the segments is in the list of marked for delete then update the status.
            if (segmentsToBeDeleted.contains(new Segment(detail.getLoadName()))) {
              detail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
            } else if (segmentFilesTobeUpdated
                .contains(Segment.toSegment(detail.getLoadName(), null))) {
              detail.setSegmentFile(
                  detail.getLoadName() + "_" + newMetaEntry.getUpdateStatusFileName()
                      + CarbonTablePath.SEGMENT_EXT);
            } else if (isUpdateStatusRequired && detail.getLoadName().equalsIgnoreCase("0")
                && !StringUtils.isBlank(uuid)) {
              detail.setUpdateStatusFileName(CarbonUpdateUtil.getUpdateStatusFileName(uuid));
            }
          }
          SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetails
              .toArray(new LoadMetadataDetails[0]));
          status = true;
        } else {
          LOGGER.error("Not able to acquire the lock for Table status updation for table " + loadModel
              .getDatabaseName() + "." + loadModel.getTableName());
        }
      } finally {
        if (carbonLock.unlock()) {
          LOGGER.info(
              "Table unlocked successfully after table status updation" + loadModel.getDatabaseName()
                  + "." + loadModel.getTableName());
        } else {
          LOGGER.error(
              "Unable to unlock Table lock for table" + loadModel.getDatabaseName() + "." + loadModel
                  .getTableName() + " during table status updation");
        }
      }
      return status;
    }
  
    /**
     * Method to create new entry for load in table status file
     *
     * @param loadMetadataDetails
     * @param loadStatus
     * @param loadStartTime
     * @param addLoadEndTime
     */
    public static void populateNewLoadMetaEntry(LoadMetadataDetails loadMetadataDetails,
        SegmentStatus loadStatus, long loadStartTime, boolean addLoadEndTime) {
      if (addLoadEndTime) {
        long loadEndDate = CarbonUpdateUtil.readCurrentTime();
        loadMetadataDetails.setLoadEndTime(loadEndDate);
      }
      loadMetadataDetails.setSegmentStatus(loadStatus);
      loadMetadataDetails.setLoadStartTime(loadStartTime);
    }
  
    public static boolean isValidEscapeSequence(String escapeChar) {
      return escapeChar.equalsIgnoreCase(NEW_LINE.getName()) ||
          escapeChar.equalsIgnoreCase(CARRIAGE_RETURN.getName()) ||
          escapeChar.equalsIgnoreCase(TAB.getName()) ||
          escapeChar.equalsIgnoreCase(BACKSPACE.getName());
    }
  
    public static boolean isValidBinaryDecoder(String binaryDecoderChar) {
      return CarbonLoadOptionConstants.CARBON_OPTIONS_BINARY_DECODER_BASE64.equalsIgnoreCase(
          binaryDecoderChar) ||
          CarbonLoadOptionConstants.CARBON_OPTIONS_BINARY_DECODER_HEX.equalsIgnoreCase(
              binaryDecoderChar) ||
          StringUtils.isBlank(binaryDecoderChar);
    }
  
    public static String getEscapeChar(String escapeCharacter) {
      if (escapeCharacter.equalsIgnoreCase(NEW_LINE.getName())) {
        return NEW_LINE.getEscapeChar();
      } else if (escapeCharacter.equalsIgnoreCase(BACKSPACE.getName())) {
        return BACKSPACE.getEscapeChar();
      } else if (escapeCharacter.equalsIgnoreCase(TAB.getName())) {
        return TAB.getEscapeChar();
      } else if (escapeCharacter.equalsIgnoreCase(CARRIAGE_RETURN.getName())) {
        return CARRIAGE_RETURN.getEscapeChar();
      }
      return escapeCharacter;
    }
  
    public static void readAndUpdateLoadProgressInTableMeta(CarbonLoadModel model,
        boolean insertOverwrite, String uuid) throws IOException {
      LoadMetadataDetails newLoadMetaEntry = new LoadMetadataDetails();
      SegmentStatus status = SegmentStatus.INSERT_IN_PROGRESS;
      if (insertOverwrite) {
        status = SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS;
      }
  
      // reading the start time of data load.
      if (model.getFactTimeStamp() == 0) {
        long loadStartTime = CarbonUpdateUtil.readCurrentTime();
        model.setFactTimeStamp(loadStartTime);
      }
      CarbonLoaderUtil
          .populateNewLoadMetaEntry(newLoadMetaEntry, status, model.getFactTimeStamp(), false);
  
      boolean entryAdded = CarbonLoaderUtil
          .recordNewLoadMetadata(newLoadMetaEntry, model, true, insertOverwrite, uuid, false);
      if (!entryAdded) {
        throw new IOException("Dataload failed due to failure in table status updation for "
            + model.getTableName());
      }
    }
  
    public static void readAndUpdateLoadProgressInTableMeta(CarbonLoadModel model,
        boolean insertOverwrite) throws IOException {
      readAndUpdateLoadProgressInTableMeta(model, insertOverwrite, "");
    }
  
    /**
     * This method will update the load failure entry in the table status file
     */
    public static void updateTableStatusForFailure(CarbonLoadModel model, String uuid)
        throws IOException {
      // in case if failure the load status should be "Marked for delete" so that it will be taken
      // care during clean up
      SegmentStatus loadStatus = SegmentStatus.MARKED_FOR_DELETE;
      // always the last entry in the load metadata details will be the current load entry
      LoadMetadataDetails loadMetaEntry = model.getCurrentLoadMetadataDetail();
      if (loadMetaEntry == null) {
        return;
      }
      CarbonLoaderUtil
          .populateNewLoadMetaEntry(loadMetaEntry, loadStatus, model.getFactTimeStamp(), true);
      boolean entryAdded = CarbonLoaderUtil.recordNewLoadMetadata(
          loadMetaEntry, model, false, false, uuid, false);
      if (!entryAdded) {
        throw new IOException(
            "Failed to update failure entry in table status for " + model.getTableName());
      }
    }
  
    /**
     * This method will update the load failure entry in the table status file with empty uuid.
     */
    public static void updateTableStatusForFailure(CarbonLoadModel model)
        throws IOException {
      updateTableStatusForFailure(model, "");
    }
  
    /**
     * This method will divide the blocks among the tasks of the nodes as per the data locality
     *
     * @param blockInfos
     * @param noOfNodesInput -1 if number of nodes has to be decided
     *                       based on block location information
     * @param parallelism    total no of tasks to execute in parallel
     * @return
     */
    public static Map<String, List<List<Distributable>>> nodeBlockTaskMapping(
        List<Distributable> blockInfos, int noOfNodesInput, int parallelism,
        List<String> activeNode) {
      Map<String, List<Distributable>> mapOfNodes =
          CarbonLoaderUtil.nodeBlockMapping(blockInfos, noOfNodesInput, activeNode,
              BlockAssignmentStrategy.BLOCK_NUM_FIRST, null);
      int taskPerNode = parallelism / mapOfNodes.size();
      //assigning non zero value to noOfTasksPerNode
      int noOfTasksPerNode = taskPerNode == 0 ? 1 : taskPerNode;
      // divide the blocks of a node among the tasks of the node.
      return assignBlocksToTasksPerNode(mapOfNodes, noOfTasksPerNode);
    }
  
    /**
     * This method will divide the blocks among the nodes as per the data locality
     * @param activeNodes List of all the active nodes running in cluster, based on these and the
     *                    actual nodes, where blocks are present, the mapping is done
     * @param blockInfos
     * @return
     */
    public static Map<String, List<Distributable>> nodeBlockMapping(List<Distributable> blockInfos,
        int noOfNodesInput, List<String> activeNodes) {
      return nodeBlockMapping(blockInfos, noOfNodesInput, activeNodes,
          BlockAssignmentStrategy.BLOCK_NUM_FIRST, null);
    }
  
    /**
     * This method will divide the blocks among the nodes as per the data locality
     *
     * @param blockInfos
     * @return
     */
    public static Map<String, List<Distributable>> nodeBlockMapping(List<Distributable> blockInfos) {
      // -1 if number of nodes has to be decided based on block location information
      return nodeBlockMapping(blockInfos, -1, null,
          BlockAssignmentStrategy.BLOCK_NUM_FIRST, null);
    }
  
    /**
     * This method will divide the blocks among the nodes as per the data locality
     *
     * @param blockInfos blocks
     * @param numOfNodesInput -1 if number of nodes has to be decided
     *                       based on block location information
     * @param blockAssignmentStrategy strategy used to assign blocks
     * @param expectedMinSizePerNode the property load_min_size_inmb specified by the user
     * @return a map that maps node to blocks
     */
    public static Map<String, List<Distributable>> nodeBlockMapping(
        List<Distributable> blockInfos, int numOfNodesInput, List<String> activeNodes,
        BlockAssignmentStrategy blockAssignmentStrategy, String expectedMinSizePerNode) {
      ArrayList<NodeMultiBlockRelation> rtnNode2Blocks = new ArrayList<>();
  
      Set<Distributable> uniqueBlocks = new HashSet<>(blockInfos);
      ArrayList<NodeMultiBlockRelation> originNode2Blocks = createNode2BlocksMapping(blockInfos);
      Set<String> nodes = new HashSet<>(originNode2Blocks.size());
      for (NodeMultiBlockRelation relation : originNode2Blocks) {
        nodes.add(relation.getNode());
      }
  
      int numOfNodes = (-1 == numOfNodesInput) ? nodes.size() : numOfNodesInput;
      if (null != activeNodes) {
        numOfNodes = activeNodes.size();
      }
  
      // calculate the average expected size for each node
      long sizePerNode = 0;
      long totalFileSize = 0;
      if (BlockAssignmentStrategy.BLOCK_NUM_FIRST == blockAssignmentStrategy) {
        if (blockInfos.size() > 0) {
          sizePerNode = blockInfos.size() / numOfNodes;
        }
        sizePerNode = sizePerNode <= 0 ? 1 : sizePerNode;
      } else if (BlockAssignmentStrategy.BLOCK_SIZE_FIRST == blockAssignmentStrategy
          || BlockAssignmentStrategy.NODE_MIN_SIZE_FIRST == blockAssignmentStrategy) {
        for (Distributable blockInfo : uniqueBlocks) {
          totalFileSize += ((TableBlockInfo) blockInfo).getBlockLength();
        }
        sizePerNode = totalFileSize / numOfNodes;
      }
  
      // if enable to control the minimum amount of input data for each node
      if (BlockAssignmentStrategy.NODE_MIN_SIZE_FIRST == blockAssignmentStrategy) {
        long expectedMinSizePerNodeInt = 0;
        // validate the property load_min_size_inmb specified by the user
        if (CarbonUtil.validateValidIntType(expectedMinSizePerNode)) {
          expectedMinSizePerNodeInt = Integer.parseInt(expectedMinSizePerNode);
        } else {
          LOGGER.warn("Invalid load_min_size_inmb value found: " + expectedMinSizePerNode
              + ", only int value greater than 0 is supported.");
          expectedMinSizePerNodeInt = Integer.parseInt(
              CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB_DEFAULT);
        }
        // If the average expected size for each node greater than load min size,
        // then fall back to default strategy
        if (expectedMinSizePerNodeInt * 1024 * 1024 < sizePerNode) {
          if (CarbonProperties.getInstance().isLoadSkewedDataOptimizationEnabled()) {
            blockAssignmentStrategy = BlockAssignmentStrategy.BLOCK_SIZE_FIRST;
          } else {
            blockAssignmentStrategy = BlockAssignmentStrategy.BLOCK_NUM_FIRST;
            // fall back to BLOCK_NUM_FIRST strategy need to reset
            // the average expected size for each node
            if (numOfNodes == 0) {
              sizePerNode = 1;
            } else {
              sizePerNode = blockInfos.size() / numOfNodes;
              sizePerNode = sizePerNode <= 0 ? 1 : sizePerNode;
            }
          }
          LOGGER.info("Specified minimum data size to load is less than the average size "
              + "for each node, fallback to default strategy" + blockAssignmentStrategy);
        } else {
          sizePerNode = expectedMinSizePerNodeInt;
        }
      }
  
      if (BlockAssignmentStrategy.NODE_MIN_SIZE_FIRST == blockAssignmentStrategy) {
        // assign blocks to each node ignore data locality
        assignBlocksIgnoreDataLocality(rtnNode2Blocks, sizePerNode, uniqueBlocks, activeNodes);
      } else {
        // assign blocks to each node
        assignBlocksByDataLocality(rtnNode2Blocks, sizePerNode, uniqueBlocks, originNode2Blocks,
            activeNodes, blockAssignmentStrategy);
      }
  
      // if any blocks remain then assign them to nodes in round robin.
      assignLeftOverBlocks(rtnNode2Blocks, uniqueBlocks, sizePerNode, activeNodes,
          blockAssignmentStrategy);
  
      // convert
      Map<String, List<Distributable>> rtnNodeBlocksMap =
          new HashMap<String, List<Distributable>>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      for (NodeMultiBlockRelation relation : rtnNode2Blocks) {
        rtnNodeBlocksMap.put(relation.getNode(), relation.getBlocks());
      }
      return rtnNodeBlocksMap;
    }
  
    /**
     * Assigning the blocks of a node to tasks.
     *
     * @param nodeBlocksMap nodeName to list of blocks mapping
     * @param noOfTasksPerNode
     * @return
     */
    private static Map<String, List<List<Distributable>>> assignBlocksToTasksPerNode(
        Map<String, List<Distributable>> nodeBlocksMap, int noOfTasksPerNode) {
      Map<String, List<List<Distributable>>> outputMap =
          new HashMap<String, List<List<Distributable>>>(
              CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  
      // for each node
      for (Map.Entry<String, List<Distributable>> eachNode : nodeBlocksMap.entrySet()) {
  
        List<Distributable> blockOfEachNode = eachNode.getValue();
        //sorting the block so same block will be give to same executor
        Collections.sort(blockOfEachNode);
        // create the task list for each node.
        createTaskListForNode(outputMap, noOfTasksPerNode, eachNode.getKey());
  
        // take all the block of node and divide it among the tasks of a node.
        divideBlockToTasks(outputMap, eachNode.getKey(), blockOfEachNode);
      }
  
      return outputMap;
    }
  
    /**
     * This will divide the blocks of a node to tasks of the node.
     *
     * @param outputMap
     * @param key
     * @param blockOfEachNode
     */
    private static void divideBlockToTasks(Map<String, List<List<Distributable>>> outputMap,
        String key, List<Distributable> blockOfEachNode) {
  
      List<List<Distributable>> taskLists = outputMap.get(key);
      int tasksOfNode = taskLists.size();
      int i = 0;
      for (Distributable block : blockOfEachNode) {
  
        taskLists.get(i % tasksOfNode).add(block);
        i++;
      }
  
    }
  
    /**
     * This will create the empty list for each task of a node.
     *
     * @param outputMap
     * @param noOfTasksPerNode
     * @param key
     */
    private static void createTaskListForNode(Map<String, List<List<Distributable>>> outputMap,
        int noOfTasksPerNode, String key) {
      List<List<Distributable>> nodeTaskList =
          new ArrayList<List<Distributable>>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      for (int i = 0; i < noOfTasksPerNode; i++) {
        List<Distributable> eachTask =
            new ArrayList<Distributable>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
        nodeTaskList.add(eachTask);
  
      }
      outputMap.put(key, nodeTaskList);
  
    }
  
    /**
     * If any left over data blocks are present then assign those to nodes in round robin way. This
     * will not obey the data locality.
     */
    private static void assignLeftOverBlocks(ArrayList<NodeMultiBlockRelation> outputMap,
        Set<Distributable> leftOverBlocks, long expectedSizePerNode, List<String> activeNodes,
        BlockAssignmentStrategy blockAssignmentStrategy) {
      Map<String, Integer> node2Idx = new HashMap<>(outputMap.size());
      for (int idx = 0; idx < outputMap.size(); idx++) {
        node2Idx.put(outputMap.get(idx).getNode(), idx);
      }
  
      // iterate all the nodes and try to allocate blocks to the nodes
      if (activeNodes != null) {
        for (String activeNode : activeNodes) {
          if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Second assignment iteration: assign for executor: " + activeNode);
          }
  
          Integer idx;
          List<Distributable> blockLst;
          if (node2Idx.containsKey(activeNode)) {
            idx = node2Idx.get(activeNode);
            blockLst = outputMap.get(idx).getBlocks();
          } else {
            idx = node2Idx.size();
            blockLst = new ArrayList<Distributable>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
          }
          populateBlocks(leftOverBlocks, expectedSizePerNode, blockLst, blockAssignmentStrategy);
  
          if (!node2Idx.containsKey(activeNode) && blockLst.size() > 0) {
            outputMap.add(idx, new NodeMultiBlockRelation(activeNode, blockLst));
            node2Idx.put(activeNode, idx);
          }
        }
      } else {
        for (NodeMultiBlockRelation entry : outputMap) {
          List<Distributable> blockLst = entry.getBlocks();
          populateBlocks(leftOverBlocks, expectedSizePerNode, blockLst, blockAssignmentStrategy);
        }
      }
  
      // if there is still blocks left, allocate them in round robin manner to each nodes
      assignBlocksUseRoundRobin(outputMap, leftOverBlocks, blockAssignmentStrategy);
    }
  
    /**
     * assign remaining blocks to nodes
     *
     * @param remainingBlocks blocks to be allocated
     * @param expectedSizePerNode expected size for each node
     * @param blockLst destination for the blocks to be allocated
     * @param blockAssignmentStrategy block assignment stretegy
     */
    private static void populateBlocks(Set<Distributable> remainingBlocks,
        long expectedSizePerNode, List<Distributable> blockLst,
        BlockAssignmentStrategy blockAssignmentStrategy) {
      switch (blockAssignmentStrategy) {
        case BLOCK_NUM_FIRST:
          populateBlocksByNum(remainingBlocks, expectedSizePerNode, blockLst);
          break;
        case BLOCK_SIZE_FIRST:
        case NODE_MIN_SIZE_FIRST:
          populateBlocksBySize(remainingBlocks, expectedSizePerNode, blockLst);
          break;
        default:
          throw new IllegalArgumentException(
              "Unsupported block assignment strategy: " + blockAssignmentStrategy);
      }
    }
  
    /**
     * Taken N number of distributable blocks from {@param remainingBlocks} and add them to output
     * {@param blockLst}. After added, the total number of {@param blockLst} is less
     * than {@param expectedSizePerNode}.
     */
    private static void populateBlocksByNum(Set<Distributable> remainingBlocks,
        long expectedSizePerNode, List<Distributable> blockLst) {
      Iterator<Distributable> blocks = remainingBlocks.iterator();
      // if the node is already having the per block nodes then avoid assign the extra blocks
      if (blockLst.size() == expectedSizePerNode) {
        return;
      }
      while (blocks.hasNext()) {
        Distributable block = blocks.next();
        blockLst.add(block);
        blocks.remove();
        if (blockLst.size() >= expectedSizePerNode) {
          break;
        }
      }
    }
  
    /**
     * Taken N number of distributable blocks from {@param remainingBlocks} and add them to output
     * {@param blockLst}. After added, the total accumulated block size of {@param blockLst}
     * is less than {@param expectedSizePerNode}.
     */
    private static void populateBlocksBySize(Set<Distributable> remainingBlocks,
        long expectedSizePerNode, List<Distributable> blockLst) {
      Iterator<Distributable> blocks = remainingBlocks.iterator();
      //if the node is already having the avg node size then avoid assign the extra blocks
      long fileSize = 0;
      for (Distributable block : blockLst) {
        fileSize += ((TableBlockInfo) block).getBlockLength();
      }
      if (fileSize >= expectedSizePerNode) {
        LOGGER.debug("Capacity is full, skip allocate blocks on this node");
        return;
      }
  
      while (blocks.hasNext()) {
        Distributable block = blocks.next();
        long thisBlockSize = ((TableBlockInfo) block).getBlockLength();
        if (fileSize < expectedSizePerNode) {
          // `fileSize==0` means there are no blocks assigned to this node before
          if (fileSize == 0 || fileSize + thisBlockSize <= expectedSizePerNode * 1.1D) {
            blockLst.add(block);
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug("Second Assignment iteration: "
                  + ((TableBlockInfo) block).getFilePath() + "-"
                  + ((TableBlockInfo) block).getBlockLength() + "-->currentNode");
            }
            fileSize += thisBlockSize;
            blocks.remove();
          }
        } else {
          break;
        }
      }
    }
  
    /**
     * allocate the blocks in round robin manner
     */
    private static void assignBlocksUseRoundRobin(ArrayList<NodeMultiBlockRelation> node2Blocks,
        Set<Distributable> remainingBlocks, BlockAssignmentStrategy blockAssignmentStrategy) {
      switch (blockAssignmentStrategy) {
        case BLOCK_NUM_FIRST:
          roundRobinAssignBlocksByNum(node2Blocks, remainingBlocks);
          break;
        case BLOCK_SIZE_FIRST:
        case NODE_MIN_SIZE_FIRST:
          roundRobinAssignBlocksBySize(node2Blocks, remainingBlocks);
          break;
        default:
          throw new IllegalArgumentException("Unsupported block assignment strategy: "
              + blockAssignmentStrategy);
      }
    }
  
    private static void roundRobinAssignBlocksByNum(ArrayList<NodeMultiBlockRelation> outputMap,
        Set<Distributable> remainingBlocks) {
      for (NodeMultiBlockRelation relation: outputMap) {
        Iterator<Distributable> blocks = remainingBlocks.iterator();
        if (blocks.hasNext()) {
          Distributable block = blocks.next();
          List<Distributable> blockLst = relation.getBlocks();
          blockLst.add(block);
          blocks.remove();
        }
      }
    }
  
    private static void roundRobinAssignBlocksBySize(ArrayList<NodeMultiBlockRelation> outputMap,
        Set<Distributable> remainingBlocks) {
      Iterator<Distributable> blocks = remainingBlocks.iterator();
      while (blocks.hasNext()) {
        // sort the allocated node-2-blocks in ascending order, the total data size of first one is
        // the smallest, so we assign this block to it.
        Collections.sort(outputMap, NodeMultiBlockRelation.DATA_SIZE_ASC_COMPARATOR);
        Distributable block = blocks.next();
        List<Distributable> blockLst = outputMap.get(0).getBlocks();
        blockLst.add(block);
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("RoundRobin assignment iteration: "
              + ((TableBlockInfo) block).getFilePath() + "-"
              + ((TableBlockInfo) block).getBlockLength() + "-->" + outputMap.get(0).getNode());
        }
        blocks.remove();
      }
    }
  
    /**
     * allocate distributable blocks to nodes based on data locality
     */
    private static void assignBlocksByDataLocality(
        ArrayList<NodeMultiBlockRelation> outputNode2Blocks,
        long expectedSizePerNode, Set<Distributable> remainingBlocks,
        List<NodeMultiBlockRelation> inputNode2Blocks, List<String> activeNodes,
        BlockAssignmentStrategy blockAssignmentStrategy) {
      if (BlockAssignmentStrategy.BLOCK_SIZE_FIRST == blockAssignmentStrategy) {
        // sort nodes based on data size of all blocks per node, so that nodes having bigger size
        // are assigned first
        Collections.sort(inputNode2Blocks, NodeMultiBlockRelation.DATA_SIZE_DESC_COMPARATOR);
      } else {
        // sort nodes based on number of blocks per node, so that nodes having lesser blocks
        // are assigned first
        Collections.sort(inputNode2Blocks);
      }
  
      Map<String, Integer> node2Idx = new HashMap<>();
      for (NodeMultiBlockRelation nodeMultiBlockRelation : inputNode2Blocks) {
        String nodeName = nodeMultiBlockRelation.getNode();
        // assign the block to the node only if the node is active
        String activeNode = nodeName;
        if (null != activeNodes) {
          activeNode = getActiveNode(activeNodes, nodeName);
          if (null == activeNode) {
            continue;
          }
        }
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("First Assignment iteration: assign for executor: " + activeNode);
        }
  
        List<Distributable> blocksInThisNode = nodeMultiBlockRelation.getBlocks();
        if (BlockAssignmentStrategy.BLOCK_SIZE_FIRST == blockAssignmentStrategy) {
          // sort blocks based on block size, so that bigger blocks will be assigned first
          Collections.sort(blocksInThisNode, TableBlockInfo.DATA_SIZE_DESC_COMPARATOR);
        }
  
        long nodeCapacity = 0;
        // loop thru blocks of each Node
        for (Distributable block : nodeMultiBlockRelation.getBlocks()) {
          if (!remainingBlocks.contains(block)) {
            // this block has been added before
            continue;
          }
          // this is the first time to add block to this node, initialize it
          if (!node2Idx.containsKey(activeNode)) {
            Integer idx = node2Idx.size();
            outputNode2Blocks.add(idx, new NodeMultiBlockRelation(activeNode,
                new ArrayList<Distributable>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE)));
            node2Idx.put(activeNode, idx);
          }
  
          // assign this block to this node if node has capacity left
          if (BlockAssignmentStrategy.BLOCK_NUM_FIRST == blockAssignmentStrategy) {
            if (nodeCapacity < expectedSizePerNode) {
              Integer idx = node2Idx.get(activeNode);
              List<Distributable> infos = outputNode2Blocks.get(idx).getBlocks();
              infos.add(block);
              nodeCapacity++;
              if (LOGGER.isDebugEnabled()) {
                try {
                  LOGGER.debug("First Assignment iteration: block("
                      + StringUtils.join(block.getLocations(), ", ")
                      + ")-->" + activeNode);
                } catch (IOException e) {
                  LOGGER.error(e.getMessage(), e);
                }
              }
              remainingBlocks.remove(block);
            } else {
              // No need to continue loop as node is full
              break;
            }
          } else if (BlockAssignmentStrategy.BLOCK_SIZE_FIRST == blockAssignmentStrategy) {
            long thisBlockSize = ((TableBlockInfo) block).getBlockLength();
            // `nodeCapacity == 0` means that there is a huge block that already exceed the
            // `expectedSize` of the node, so we have to assign it to some node, otherwise it will
            // be assigned in the last RoundRobin iteration.
            if (nodeCapacity == 0 || nodeCapacity < expectedSizePerNode) {
              if (nodeCapacity == 0 || nodeCapacity + thisBlockSize <= expectedSizePerNode * 1.05D) {
                Integer idx = node2Idx.get(activeNode);
                List<Distributable> blocks = outputNode2Blocks.get(idx).getBlocks();
                blocks.add(block);
                nodeCapacity += thisBlockSize;
                if (LOGGER.isDebugEnabled()) {
                  LOGGER.debug(
                      "First Assignment iteration: " + ((TableBlockInfo) block).getFilePath() + '-'
                          + ((TableBlockInfo) block).getBlockLength() + "-->" + activeNode);
                }
                remainingBlocks.remove(block);
              }
              // this block is too big for current node and there are still capacity left
              // for small files, so continue to allocate block on this node in next iteration.
            } else {
              // No need to continue loop as node is full
              break;
            }
          } else {
            throw new IllegalArgumentException(
                "Unsupported block assignment strategy: " + blockAssignmentStrategy);
          }
        }
      }
    }
  
    /**
     * allocate distributable blocks to nodes based on ignore data locality
     */
    private static void assignBlocksIgnoreDataLocality(
            ArrayList<NodeMultiBlockRelation> outputNode2Blocks,
            long expectedSizePerNode, Set<Distributable> remainingBlocks,
            List<String> activeNodes) {
      // get all blocks
      Set<Distributable> uniqueBlocks = new HashSet<>(remainingBlocks);
      // shuffle activeNodes ignore data locality
      List<String> shuffleNodes  = new ArrayList<>(activeNodes);
      Collections.shuffle(shuffleNodes);
  
      for (String activeNode : shuffleNodes) {
        long nodeCapacity = 0;
        NodeMultiBlockRelation nodeBlock = new NodeMultiBlockRelation(activeNode,
            new ArrayList<Distributable>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE));
        // loop thru blocks of each Node
        for (Distributable block : uniqueBlocks) {
          if (!remainingBlocks.contains(block)) {
            // this block has been added before
            continue;
          }
  
          long thisBlockSize = ((TableBlockInfo) block).getBlockLength();
          if (nodeCapacity == 0
              || nodeCapacity + thisBlockSize <= expectedSizePerNode * 1024 * 1024) {
            nodeBlock.getBlocks().add(block);
            nodeCapacity += thisBlockSize;
            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug(
                  "First Assignment iteration: " + ((TableBlockInfo) block).getFilePath() + '-'
                      + ((TableBlockInfo) block).getBlockLength() + "-->" + activeNode);
            }
            remainingBlocks.remove(block);
            // this block is too big for current node and there are still capacity left
            // for small files, so continue to allocate block on this node in next iteration.
          } else {
            // No need to continue loop as node is full
            break;
          }
        }
        if (nodeBlock.getBlocks().size() != 0) {
          outputNode2Blocks.add(nodeBlock);
        }
      }
    }
  
    /**
     * method validates whether the node is active or not.
     *
     * @param activeNode
     * @param nodeName
     * @return hostName or hostAddress if node is active
     */
    private static String getActiveNode(List<String> activeNode, String nodeName) {
      try {
        boolean isActiveNode = activeNode.contains(nodeName);
        if (isActiveNode) {
          return nodeName;
        }
        //if localhost then retrieve the localhost name then do the check
        else if (nodeName.equals("localhost")) {
          String hostName = InetAddress.getLocalHost().getHostName();
          isActiveNode = activeNode.contains(hostName);
          if (isActiveNode) {
            return hostName;
          }
        } else {
          String hostAddress = InetAddress.getByName(nodeName).getHostAddress();
          isActiveNode = activeNode.contains(hostAddress);
          if (isActiveNode) {
            return hostAddress;
          }
        }
      } catch (UnknownHostException ue) {
        return null;
      }
      return null;
    }
  
    /**
     * Create node to blocks mapping
     *
     * @param blockInfos input block info
     */
    private static ArrayList<NodeMultiBlockRelation> createNode2BlocksMapping(
        List<Distributable> blockInfos) {
      Map<String, Integer> node2Idx = new HashMap<>();
      ArrayList<NodeMultiBlockRelation> node2Blocks = new ArrayList<>();
  
      for (Distributable blockInfo : blockInfos) {
        try {
          for (final String eachNode : blockInfo.getLocations()) {
            if (node2Idx.containsKey(eachNode)) {
              Integer idx = node2Idx.get(eachNode);
              List<Distributable> blocks = node2Blocks.get(idx).getBlocks();
              blocks.add(blockInfo);
            } else {
              // add blocks to this node for the first time
              Integer idx = node2Idx.size();
              List<Distributable> blocks = new ArrayList<>();
              blocks.add(blockInfo);
              node2Blocks.add(idx, new NodeMultiBlockRelation(eachNode, blocks));
              node2Idx.put(eachNode, idx);
            }
          }
        } catch (IOException e) {
          throw new RuntimeException("error getting location of block: " + blockInfo.toString(), e);
        }
      }
  
      return node2Blocks;
    }
  
    /**
     * This method will get the store location for the given path, segment id and partition id
     */
    public static void checkAndCreateCarbonDataLocation(String segmentId, CarbonTable carbonTable) {
      String segmentFolder = CarbonTablePath.getSegmentPath(
          carbonTable.getTablePath(), segmentId);
      CarbonUtil.checkAndCreateFolder(segmentFolder);
    }
  
    /*
     * This method will add data size and index size into tablestatus for each segment. And also
     * returns the size of the segment.
     */
    public static Long addDataIndexSizeIntoMetaEntry(LoadMetadataDetails loadMetadataDetails,
        String segmentId, CarbonTable carbonTable) throws IOException {
      Map<String, Long> dataIndexSize = CarbonUtil.getDataSizeAndIndexSize(
          carbonTable.getTablePath(),
          new Segment(segmentId, loadMetadataDetails.getSegmentFile()));
      Long dataSize = dataIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_DATA_SIZE);
      loadMetadataDetails.setDataSize(String.valueOf(dataSize));
      Long indexSize = dataIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_INDEX_SIZE);
      loadMetadataDetails.setIndexSize(String.valueOf(indexSize));
      return dataSize + indexSize;
    }
  
    public static void addIndexSizeIntoMetaEntry(LoadMetadataDetails loadMetadataDetails,
        String segmentId, CarbonTable carbonTable) throws IOException {
      Segment segment = new Segment(segmentId, loadMetadataDetails.getSegmentFile());
      if (segment.getSegmentFileName() != null) {
        SegmentFileStore fileStore =
            new SegmentFileStore(carbonTable.getTablePath(), segment.getSegmentFileName());
        if (fileStore.getLocationMap() != null) {
          fileStore.readIndexFiles(FileFactory.getConfiguration());
          long carbonIndexSize = CarbonUtil.getCarbonIndexSize(fileStore, fileStore.getLocationMap());
          loadMetadataDetails.setIndexSize(String.valueOf(carbonIndexSize));
        }
      }
    }
  
    /**
     * Merge index files with in the segment of partitioned table
     *
     * @param table
     * @param segmentId
     * @param uuid
     * @return
     */
    public static String mergeIndexFilesInPartitionedSegment(CarbonTable table, String segmentId,
        String uuid, String partitionPath, boolean isOldStoreIndexFilesPresent) {
      String tablePath = table.getTablePath();
      return new CarbonIndexFileMergeWriter(table)
          .mergeCarbonIndexFilesOfSegment(segmentId, uuid, tablePath, partitionPath,
              isOldStoreIndexFilesPresent);
    }
  
    public static SegmentFileStore.FolderDetails mergeIndexFilesInPartitionedTempSegment(
        CarbonTable table, String segmentId, String partitionPath, List<String> partitionInfo,
        String uuid, String tempFolderPath, String currPartitionSpec) throws IOException {
      String tablePath = table.getTablePath();
      return new CarbonIndexFileMergeWriter(table)
          .mergeCarbonIndexFilesOfSegment(segmentId, tablePath, partitionPath, partitionInfo, uuid,
              tempFolderPath, currPartitionSpec);
    }
  
    /**
     * Merge index files of segment with external path.
     */
    public static String mergeIndexFilesInAddLoadSegment(CarbonTable table, String segmentId,
        String segmentPath, String uuid) {
      try {
        return new CarbonIndexFileMergeWriter(table)
            .writeMergeIndexFileBasedOnSegmentFolder(null, false, segmentPath, segmentId, uuid,
                false);
      } catch (IOException e) {
        String message =
            "Failed to merge index files in path: " + segmentPath + ". " + e.getMessage();
        LOGGER.error(message);
        throw new RuntimeException(message, e);
      }
    }
  
    private static void deleteFiles(List<String> filesToBeDeleted) throws IOException {
      for (String filePath : filesToBeDeleted) {
        FileFactory.deleteFile(filePath);
      }
    }
  
    /**
     * Update specified segment status for load to MarkedForDelete in case of failure
     */
    public static void updateTableStatusInCaseOfFailure(String loadName,
        CarbonTable carbonTable, SegmentStatus status) throws IOException {
      updateTableStatusInCaseOfFailure(loadName,
          carbonTable.getAbsoluteTableIdentifier(),
          carbonTable.getTableName(),
          carbonTable.getDatabaseName(),
          carbonTable.getTablePath(),
          carbonTable.getMetadataPath(),
          status,
          carbonTable.getTableStatusVersion());
    }
  
    /**
     * Update specified segment status for load to MarkedForDelete in case of failure
     */
    public static void updateTableStatusInCaseOfFailure(String loadName,
        AbsoluteTableIdentifier absoluteTableIdentifier, String tableName, String databaseName,
        String tablePath, String metaDataPath, SegmentStatus status, String version)
        throws IOException {
      SegmentStatusManager segmentStatusManager =
          new SegmentStatusManager(absoluteTableIdentifier, version);
      ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
      try {
        if (carbonLock.lockWithRetries()) {
          LOGGER.info("Acquired lock for table" + databaseName + "." + tableName
              + " for table status updation");
          LoadMetadataDetails[] loadMetadataDetails =
              SegmentStatusManager.readLoadMetadata(metaDataPath, version);
          boolean ifTableStatusUpdateRequired = false;
          for (LoadMetadataDetails loadMetadataDetail : loadMetadataDetails) {
            if (loadMetadataDetail.getSegmentStatus() == status && loadName
                .equalsIgnoreCase(loadMetadataDetail.getLoadName())) {
              loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
              ifTableStatusUpdateRequired = true;
              break;
            }
          }
          if (ifTableStatusUpdateRequired) {
            SegmentStatusManager
                .writeLoadDetailsIntoFile(CarbonTablePath.getTableStatusFilePath(tablePath, version),
                    loadMetadataDetails);
          }
        } else {
          LOGGER.error(
              "Not able to acquire the lock for Table status updation for table " + databaseName + "."
                  + tableName);
        }
      } finally {
        if (carbonLock.unlock()) {
          LOGGER.info("Table unlocked successfully after table status updation" + databaseName + "."
              + tableName);
        } else {
          LOGGER.error("Unable to unlock Table lock for table" + databaseName + "." + tableName
              + " during table status updation");
        }
      }
    }
  }
  