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.core.mutate;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.stream.Collectors;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.constants.CarbonCommonConstantsInternal;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.index.Segment;
  import org.apache.carbondata.core.locks.ICarbonLock;
  import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
  import org.apache.carbondata.core.metadata.SegmentFileStore;
  import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
  import org.apache.carbondata.core.mutate.data.BlockMappingVO;
  import org.apache.carbondata.core.mutate.data.RowCountDetailsVO;
  import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
  import org.apache.carbondata.core.statusmanager.SegmentStatus;
  import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
  import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
  import org.apache.carbondata.core.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.path.CarbonTablePath;
  
  import org.apache.commons.collections.CollectionUtils;
  import org.apache.hadoop.fs.Path;
  import org.apache.log4j.Logger;
  
  /**
   * This class contains all update utility methods
   */
  public class CarbonUpdateUtil {
  
    private static final Logger LOGGER =
            LogServiceFactory.getLogService(CarbonUpdateUtil.class.getName());
  
    /**
     * returns required filed from tuple id
     *
     */
    public static String getRequiredFieldFromTID(String Tid, int index) {
      return Tid.split(CarbonCommonConstants.FILE_SEPARATOR)[index];
    }
  
    /**
     * returns required filed from tuple id
     *
     * @param Tid
     * @param tid
     * @return
     */
    public static String getRequiredFieldFromTID(String Tid, TupleIdEnum tid) {
      return Tid.split("/")[tid.getTupleIdIndex()];
    }
  
    /**
     * returns segment along with block id
     * @param Tid
     * @return
     */
    public static String getSegmentWithBlockFromTID(String Tid, boolean isPartitionTable) {
      if (isPartitionTable) {
        return getRequiredFieldFromTID(Tid, TupleIdEnum.PARTITION_SEGMENT_ID);
      }
      // this case is to check for add segment case, as now the segment id is present at first index,
      // in add segment case, it will be in second index as the blockletID is generated by adding the
      // complete external path
      // this is in case of the external segment, where the tuple id has external path with #
      // here no need to check for any path present in metadta details, as # can come in tuple id in
      // case of multiple partitions, so partition check is already present above.
      if (Tid.contains("#/")) {
        return getRequiredFieldFromTID(Tid, TupleIdEnum.EXTERNAL_SEGMENT_ID)
            + CarbonCommonConstants.FILE_SEPARATOR + getRequiredFieldFromTID(Tid,
            TupleIdEnum.EXTERNAL_BLOCK_ID);
      }
      return getRequiredFieldFromTID(Tid, TupleIdEnum.SEGMENT_ID)
          + CarbonCommonConstants.FILE_SEPARATOR + getRequiredFieldFromTID(Tid, TupleIdEnum.BLOCK_ID);
    }
  
    /**
     * Returns block path from tuple id
     */
    public static String getTableBlockPath(String tid, String tablePath, boolean isStandardTable,
        boolean isPartitionTable) {
      String partField = "0";
      // If it has segment file then part field can be appended directly to table path
      if (!isStandardTable) {
        if (isPartitionTable) {
          partField = getRequiredFieldFromTID(tid, TupleIdEnum.PARTITION_PART_ID);
          return tablePath + CarbonCommonConstants.FILE_SEPARATOR + partField.replace("#", "/");
        } else {
          return tablePath;
        }
      }
      String part = CarbonTablePath.addPartPrefix(partField);
      String segment =
              CarbonTablePath.addSegmentPrefix(getRequiredFieldFromTID(tid, TupleIdEnum.SEGMENT_ID));
      return CarbonTablePath.getFactDir(tablePath) + CarbonCommonConstants.FILE_SEPARATOR + part
              + CarbonCommonConstants.FILE_SEPARATOR + segment;
    }
  
    /**
     * returns delete delta file path
     *
     * @param blockPath
     * @param blockPath
     * @param timestamp
     * @return
     */
    public static String getDeleteDeltaFilePath(String blockPath, String blockName,
                                                String timestamp) {
      return blockPath + CarbonCommonConstants.FILE_SEPARATOR + blockName
          + CarbonCommonConstants.HYPHEN + timestamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT;
  
    }
  
    /**
     * @param updateDetailsList
     * @param table
     * @param updateStatusFileIdentifier
     * @return
     */
    public static boolean updateSegmentStatus(List<SegmentUpdateDetails> updateDetailsList,
        CarbonTable table, String updateStatusFileIdentifier, boolean isCompaction,
        boolean isForceWrite) {
      boolean status = false;
      SegmentUpdateStatusManager segmentUpdateStatusManager = new SegmentUpdateStatusManager(table);
      if (isForceWrite && !CollectionUtils.isEmpty(updateDetailsList)) {
        String segId = String.valueOf(SegmentStatusManager
            .createNewSegmentId(segmentUpdateStatusManager.getLoadMetadataDetails()));
        for (SegmentUpdateDetails detail : updateDetailsList) {
          detail.setSegmentName(segId);
        }
      }
      ICarbonLock updateLock = segmentUpdateStatusManager.getTableUpdateStatusLock();
      boolean lockStatus = false;
  
      try {
        lockStatus = updateLock.lockWithRetries();
        if (lockStatus) {
  
          // read the existing file if present and update the same.
          SegmentUpdateDetails[] oldDetails = segmentUpdateStatusManager
                  .getUpdateStatusDetails();
  
          List<SegmentUpdateDetails> oldList = new ArrayList(Arrays.asList(oldDetails));
  
          for (SegmentUpdateDetails newBlockEntry : updateDetailsList) {
            mergeSegmentUpdate(isCompaction, oldList, newBlockEntry);
          }
  
          List<SegmentUpdateDetails> updateDetailsValidSeg = new ArrayList<>();
          Set<String> loadDetailsSet = new HashSet<>();
          for (LoadMetadataDetails details : segmentUpdateStatusManager.getLoadMetadataDetails()) {
            loadDetailsSet.add(details.getLoadName());
          }
          for (SegmentUpdateDetails updateDetails : oldList) {
            if (loadDetailsSet.contains(updateDetails.getSegmentName())) {
              // we should only keep the update info of segments in table status, especially after
              // compaction and clean files some compacted segments will be removed. It can keep
              // tableupdatestatus file in small size which is good for performance.
              updateDetailsValidSeg.add(updateDetails);
            }
          }
          // In case of ForceWrite, write the segmentUpdateDetails to the tableUpdateStatus file
          // without any validation of segments.
          if (isForceWrite) {
            segmentUpdateStatusManager.writeLoadDetailsIntoFile(oldList, updateStatusFileIdentifier);
          } else {
            segmentUpdateStatusManager
                .writeLoadDetailsIntoFile(updateDetailsValidSeg, updateStatusFileIdentifier);
          }
          status = true;
        } else {
          LOGGER.error("Not able to acquire the segment update lock.");
          status = false;
        }
      } catch (IOException e) {
        status = false;
      } finally {
        if (lockStatus) {
          if (updateLock.unlock()) {
            LOGGER.info("Unlock the segment update lock successful.");
          } else {
            LOGGER.error("Not able to unlock the segment update lock.");
          }
        }
      }
      return status;
    }
  
    public static void mergeSegmentUpdate(boolean isCompaction, List<SegmentUpdateDetails> oldList,
        SegmentUpdateDetails newBlockEntry) {
      int index = oldList.indexOf(newBlockEntry);
      if (index != -1) {
        // update the element in existing list.
        SegmentUpdateDetails blockDetail = oldList.get(index);
        if (blockDetail.getDeleteDeltaStartTimestamp().isEmpty() || isCompaction) {
          blockDetail
              .setDeleteDeltaStartTimestamp(newBlockEntry.getDeleteDeltaStartTimestamp());
        }
        blockDetail.setDeleteDeltaEndTimestamp(newBlockEntry.getDeleteDeltaEndTimestamp());
        blockDetail.setSegmentStatus(newBlockEntry.getSegmentStatus());
        blockDetail.setDeletedRowsInBlock(newBlockEntry.getDeletedRowsInBlock());
        // If the start and end time is different then the delta is there in multiple files so
        // add them to the list to get the delta files easily with out listing.
        if (!blockDetail.getDeleteDeltaStartTimestamp()
            .equals(blockDetail.getDeleteDeltaEndTimestamp())) {
          blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaStartTimestamp());
          blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaEndTimestamp());
        } else {
          blockDetail.setDeltaFileStamps(null);
        }
      } else {
        // add the new details to the list.
        oldList.add(newBlockEntry);
      }
    }
  
    /**
     * Update table status
     * @param updatedSegmentsList
     * @param table
     * @param updatedTimeStamp
     * @param isTimestampUpdateRequired
     * @param segmentsToBeDeleted
     * @return
     */
    public static Map<String, String> updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
        CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
        boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted,
        String tblStatusWriteVersion) {
      return updateTableMetadataStatus(updatedSegmentsList, table, updatedTimeStamp,
          isTimestampUpdateRequired, isUpdateStatusFileUpdateRequired,
          segmentsToBeDeleted, new ArrayList<Segment>(), tblStatusWriteVersion);
    }
  
    /**
     *
     * @param updatedSegmentsList
     * @param table
     * @param updatedTimeStamp
     * @param isTimestampUpdateRequired
     * @param segmentsToBeDeleted
     * @return
     */
    public static Map<String, String> updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
        CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
        boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted,
        List<Segment> segmentFilesTobeUpdated, String uuid) {
  
      boolean status = false;
      Map<String, String> tuple = new HashMap<>();
      String metaDataFilepath = table.getMetadataPath();
      AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
      String tableStatusPath =
          CarbonTablePath.getTableStatusFilePath(identifier.getTablePath(), uuid);
      SegmentStatusManager segmentStatusManager =
          new SegmentStatusManager(identifier, table.getTableStatusVersion());
  
      ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
      boolean lockStatus = false;
      try {
        lockStatus = carbonLock.lockWithRetries();
        if (lockStatus) {
          LOGGER.info("Acquired lock for table" + table.getDatabaseName() + "." + table.getTableName()
               + " for table status update");
  
          if (uuid.isEmpty() && CarbonProperties.isTableStatusMultiVersionEnabled()) {
            String tblStatusWriteVersion = String.valueOf(System.currentTimeMillis());
            tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath(),
                tblStatusWriteVersion);
            tuple.put("tblStatusWriteVersion", tblStatusWriteVersion);
          }
          LoadMetadataDetails[] listOfLoadFolderDetailsArray =
              SegmentStatusManager.readLoadMetadata(metaDataFilepath, table.getTableStatusVersion());
          // to update table status only when required.
          boolean isUpdateRequired = false;
  
          for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
            // we are storing the link between the 2 status files in the segment 0 only.
            if (isUpdateStatusFileUpdateRequired &&
                loadMetadata.getLoadName().equalsIgnoreCase("0")) {
              loadMetadata.setUpdateStatusFileName(
                  CarbonUpdateUtil.getUpdateStatusFileName(updatedTimeStamp));
            }
  
            if (isTimestampUpdateRequired) {
              // if the segments is in the list of marked for delete then update the status.
              if (segmentsToBeDeleted.contains(new Segment(loadMetadata.getLoadName()))) {
                loadMetadata.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
                loadMetadata.setModificationOrDeletionTimestamp(Long.parseLong(updatedTimeStamp));
                isUpdateRequired = true;
              }
            }
            for (Segment segName : updatedSegmentsList) {
              if (loadMetadata.getLoadName().equalsIgnoreCase(segName.getSegmentNo())) {
                // if this call is coming from the delete delta flow then the time stamp
                // String will come empty then no need to write into table status file.
                if (isTimestampUpdateRequired) {
                  // if in case of update flow.
                  if (loadMetadata.getUpdateDeltaStartTimestamp().isEmpty()) {
                    // this means for first time it is getting updated .
                    loadMetadata.setUpdateDeltaStartTimestamp(updatedTimeStamp);
                  }
                  // update end timestamp for each time.
                  loadMetadata.setUpdateDeltaEndTimestamp(updatedTimeStamp);
                  isUpdateRequired = true;
                }
                if (segmentFilesTobeUpdated
                    .contains(Segment.toSegment(loadMetadata.getLoadName(), null))) {
                  loadMetadata.setSegmentFile(loadMetadata.getLoadName() + "_" + updatedTimeStamp
                      + CarbonTablePath.SEGMENT_EXT);
                  isUpdateRequired = true;
                }
              }
            }
          }
  
          try {
            if (isUpdateRequired || isUpdateStatusFileUpdateRequired) {
              SegmentStatusManager
                  .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray);
            }
          } catch (IOException e) {
            tuple.put("status", "false");
            return tuple;
          }
  
          status = true;
        } else {
          LOGGER.error("Not able to acquire the lock for Table status update for table " + table
                  .getDatabaseName() + "." + table.getTableName());
        }
      } finally {
        if (lockStatus) {
          if (carbonLock.unlock()) {
            LOGGER.info(
                   "Table unlocked successfully after table status update" + table.getDatabaseName()
                            + "." + table.getTableName());
          } else {
            LOGGER.error(
                    "Unable to unlock Table lock for table" + table.getDatabaseName() + "." + table
                            .getTableName() + " during table status update");
          }
        }
      }
      tuple.put("status", String.valueOf(status));
      return tuple;
  
    }
  
    /**
     * gets the file name of the update status file. by appending the latest timestamp to it.
     *
     * @param updatedTimeStamp
     * @return
     */
    public static String getUpdateStatusFileName(String updatedTimeStamp) {
      return CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME + CarbonCommonConstants.HYPHEN
              + updatedTimeStamp;
    }
  
    /**
     * This will handle the clean up cases if the update fails.
     *
     * @param table
     * @param timeStamp
     */
    public static void cleanStaleDeltaFiles(CarbonTable table, final String timeStamp) {
  
      AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
      String partitionDir = CarbonTablePath.getPartitionDir(identifier.getTablePath());
      CarbonFile file =
              FileFactory.getCarbonFile(partitionDir);
      if (!file.exists()) {
        return;
      }
      for (CarbonFile eachDir : file.listFiles()) {
        // for each dir check if the file with the delta timestamp is present or not.
        CarbonFile[] toBeDeleted = eachDir.listFiles(new CarbonFileFilter() {
          @Override
          public boolean accept(CarbonFile file) {
            String fileName = file.getName();
            return fileName.endsWith(timeStamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT);
          }
        });
        // deleting the files of a segment.
        try {
          CarbonUtil.deleteFoldersAndFilesSilent(toBeDeleted);
        } catch (IOException | InterruptedException e) {
          LOGGER.error("Exception in deleting the delta files." + e);
        }
      }
    }
  
    /**
     * returns timestamp as long value
     *
     * @param timestamp
     * @return
     */
    public static Long getTimeStampAsLong(String timestamp) {
      try {
        return Long.parseLong(timestamp);
      } catch (NumberFormatException nfe) {
        String errorMsg = "Invalid timestamp : " + timestamp;
        LOGGER.error(errorMsg);
        return null;
      }
    }
  
    /**
     * returns integer value from given string
     *
     * @param value
     * @return
     * @throws Exception
     */
    public static Integer getIntegerValue(String value) throws Exception {
      try {
        return Integer.parseInt(value);
      } catch (NumberFormatException nfe) {
        LOGGER.error("Invalid row : " + value + nfe.getLocalizedMessage());
        throw new Exception("Invalid row : " + nfe.getLocalizedMessage());
      }
    }
  
    /**
     * return only block name from completeBlockName
     *
     * @param completeBlockName
     * @return
     */
    public static String getBlockName(String completeBlockName) {
      return completeBlockName
          .substring(0, completeBlockName.lastIndexOf(CarbonCommonConstants.HYPHEN));
    }
  
    /**
     * returns segment id from segment name
     *
     * @param segmentName
     * @return
     */
    public static String getSegmentId(String segmentName) {
      return segmentName.split(CarbonCommonConstants.UNDERSCORE)[1];
    }
  
    public static long getLatestTaskIdForSegment(Segment segment, String tablePath)
        throws IOException {
      long max = 0;
      List<String> dataFiles = new ArrayList<>();
      if (segment.getSegmentFileName() != null) {
        SegmentFileStore fileStore = new SegmentFileStore(tablePath, segment.getSegmentFileName());
        fileStore.readIndexFiles(FileFactory.getConfiguration());
        Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
        List<String> dataFilePaths = new ArrayList<>();
        for (List<String> paths : indexFilesMap.values()) {
          dataFilePaths.addAll(paths);
        }
        for (String dataFilePath : dataFilePaths) {
          dataFiles.add(new Path(dataFilePath).getName());
        }
  
      } else {
        String segmentDirPath = CarbonTablePath.getSegmentPath(tablePath, segment.getSegmentNo());
        // scan all the carbondata files and get the latest task ID.
        CarbonFile segmentDir =
            FileFactory.getCarbonFile(segmentDirPath);
        CarbonFile[] carbonDataFiles = segmentDir.listFiles(new CarbonFileFilter() {
          @Override
          public boolean accept(CarbonFile file) {
            return file.getName().endsWith(CarbonCommonConstants.FACT_FILE_EXT);
          }
        });
        for (CarbonFile carbonDataFile : carbonDataFiles) {
          dataFiles.add(carbonDataFile.getName());
        }
      }
      for (String name : dataFiles) {
        long taskNumber =
            Long.parseLong(CarbonTablePath.DataFileUtil.getTaskNo(name).split("_")[0]);
        if (taskNumber > max) {
          max = taskNumber;
        }
      }
      // return max task No
      return max;
  
    }
  
    /**
     * This will tell whether the max query timeout has been expired or not.
     * @param fileTimestamp
     * @return
     */
    public static boolean isMaxQueryTimeoutExceeded(long fileTimestamp) {
      // record current time.
      long currentTime = CarbonUpdateUtil.readCurrentTime();
      int maxTime;
      try {
        maxTime = Integer.parseInt(CarbonProperties.getInstance()
            .getProperty(CarbonCommonConstants.MAX_QUERY_EXECUTION_TIME,
                Integer.toString(CarbonCommonConstants.DEFAULT_MAX_QUERY_EXECUTION_TIME)));
      } catch (NumberFormatException e) {
        maxTime = CarbonCommonConstants.DEFAULT_MAX_QUERY_EXECUTION_TIME;
      }
  
      long difference = currentTime - fileTimestamp;
  
      long minutesElapsed = (difference / (1000 * 60));
  
      return minutesElapsed >= maxTime;
  
    }
  
    private static boolean deleteInvalidFiles(CarbonFile invalidFile) {
      boolean isDeleted;
      try {
        LOGGER.info("deleting the invalid file : " + invalidFile.getName());
        CarbonUtil.deleteFoldersAndFiles(invalidFile);
        isDeleted = true;
      } catch (IOException | InterruptedException e) {
        LOGGER.error("error in clean up of invalid files." + e.getMessage(), e);
        isDeleted = false;
      }
      return isDeleted;
    }
  
    public static boolean isBlockInvalid(SegmentStatus blockStatus) {
      return blockStatus == SegmentStatus.COMPACTED || blockStatus == SegmentStatus.MARKED_FOR_DELETE;
    }
  
    /**
     * This will return the current time in millis.
     * @return
     */
    public static long readCurrentTime() {
      return System.currentTimeMillis();
    }
  
    /**
     *
     * @param details
     * @param segmentBlockCount
     */
    public static void decrementDeletedBlockCount(SegmentUpdateDetails details,
                                                  Map<String, Long> segmentBlockCount) {
  
      String segId = details.getSegmentName();
  
      segmentBlockCount.put(details.getSegmentName(), segmentBlockCount.get(segId) - 1);
  
    }
  
    /**
     *
     * @param segmentBlockCount
     * @return
     */
    public static List<Segment> getListOfSegmentsToMarkDeleted(Map<String, Long> segmentBlockCount) {
      List<Segment> segmentsToBeDeleted =
          new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  
      for (Map.Entry<String, Long> eachSeg : segmentBlockCount.entrySet()) {
  
        if (eachSeg.getValue() == 0) {
          segmentsToBeDeleted.add(new Segment(eachSeg.getKey(), (String) null));
        }
  
      }
      return segmentsToBeDeleted;
    }
  
    /**
     * Return row count of input block
     */
    public static long getRowCount(BlockMappingVO blockMappingVO, CarbonTable carbonTable) {
      if (blockMappingVO.getBlockRowCountMapping().size() == 1
          && blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT)
          != null) {
        return blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT);
      }
      SegmentUpdateStatusManager updateStatusManager =
          new SegmentUpdateStatusManager(carbonTable);
      long rowCount = 0;
      Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping();
      for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) {
        String key = blockRowEntry.getKey();
        long alreadyDeletedCount = 0;
        SegmentUpdateDetails detail = updateStatusManager.getDetailsForABlock(key);
        if (detail != null) {
          alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock());
        }
        rowCount += (blockRowEntry.getValue() - alreadyDeletedCount);
      }
      return rowCount;
    }
  
    /**
     *
     * @param blockMappingVO
     * @param segmentUpdateStatusManager
     */
    public static void createBlockDetailsMap(BlockMappingVO blockMappingVO,
                                             SegmentUpdateStatusManager segmentUpdateStatusManager) {
  
      Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping();
  
      Map<String, RowCountDetailsVO> outputMap =
              new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  
      for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) {
        String key = blockRowEntry.getKey();
        long alreadyDeletedCount = 0;
  
        SegmentUpdateDetails detail = segmentUpdateStatusManager.getDetailsForABlock(key);
  
        if (null != detail) {
  
          alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock());
  
        }
  
        RowCountDetailsVO rowCountDetailsVO =
                new RowCountDetailsVO(blockRowEntry.getValue(), alreadyDeletedCount);
        outputMap.put(key, rowCountDetailsVO);
  
      }
  
      blockMappingVO.setCompleteBlockRowDetailVO(outputMap);
  
    }
  
    /**
     *
     * @param segID
     * @param blockName
     * @return
     */
    public static String getSegmentBlockNameKey(String segID, String blockName,
        boolean isPartitionTable) {
      String blockNameWithOutPartAndBatchNo = blockName
          .substring(blockName.indexOf(CarbonCommonConstants.HYPHEN) + 1,
              blockName.lastIndexOf(CarbonTablePath.getCarbonDataExtension()))
          .replace(CarbonTablePath.BATCH_PREFIX, CarbonCommonConstants.UNDERSCORE);
      // to remove compressor name
      int index = blockNameWithOutPartAndBatchNo.lastIndexOf(CarbonCommonConstants.POINT);
      if (index != -1) {
        blockNameWithOutPartAndBatchNo = blockNameWithOutPartAndBatchNo
            .replace(blockNameWithOutPartAndBatchNo.substring(index), "");
      }
      if (isPartitionTable) {
        return blockNameWithOutPartAndBatchNo;
      } else {
        return segID + CarbonCommonConstants.FILE_SEPARATOR + blockNameWithOutPartAndBatchNo;
      }
    }
  
    /**
     * Below method will be used to get the latest delete delta file timestamp
     * @param deleteDeltaFiles
     * @return latest delete delta file time stamp
     */
    public static long getLatestDeleteDeltaTimestamp(String[] deleteDeltaFiles) {
      long latestTimestamp = 0;
      for (int i = 0; i < deleteDeltaFiles.length; i++) {
        long convertTimeStampToLong = Long.parseLong(
            CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(deleteDeltaFiles[i]));
        if (latestTimestamp < convertTimeStampToLong) {
          latestTimestamp = convertTimeStampToLong;
        }
      }
      return latestTimestamp;
    }
  
  
    /**
     * Handling of the clean up of old carbondata files, index files , delete delta,
     * update status files.
     *
     * @param table       clean up will be handled on this table.
     * @param isDryRun if clean files dryRun selected, then only size will be shown,
     *                files will not be deleted.
     */
    public static long cleanUpDeltaFiles(CarbonTable table, boolean isDryRun) throws IOException {
  
      SegmentStatusManager ssm =
          new SegmentStatusManager(table.getAbsoluteTableIdentifier(), table.getTableStatusVersion());
      LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(table.getMetadataPath(),
          table.getTableStatusVersion());
      long totalSizeDeleted = 0;
      ArrayList<CarbonFile> filesToBeDeleted = new ArrayList<>();
      SegmentUpdateStatusManager updateStatusManager = new SegmentUpdateStatusManager(table);
      SegmentUpdateDetails[] segmentUpdateDetails = updateStatusManager.getUpdateStatusDetails();
      // hold all the segments updated so that wen can check the delta files in them, ne need to
      // check the others.
      Set<String> updatedSegments = new HashSet<>();
      for (SegmentUpdateDetails updateDetails : segmentUpdateDetails) {
        updatedSegments.add(updateDetails.getSegmentName());
      }
      boolean isInvalidFile = false;
      // take the update status file name from 0th segment.
      String validUpdateStatusFile = ssm.getUpdateStatusFileName(details);
      // scan through each segment.
  
      if (table.isHivePartitionTable()) {
        List<CarbonFile> partitionList = Arrays.stream(FileFactory.getCarbonFile(table
                .getTablePath()).listFiles()).filter(partitionName -> partitionName.getName()
                .contains("=")).collect(Collectors.toList());
  
        List<CarbonFile> totalDeltaFiles = new ArrayList<>();
  
        for (CarbonFile carbonFile : partitionList) {
          totalDeltaFiles.addAll(carbonFile.listFiles(true).stream().filter(fileName -> fileName
                  .getName().endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT))
                  .collect(Collectors.toList()));
        }
        SegmentUpdateDetails[] updateDetails = updateStatusManager.readLoadMetadata();
  
        // Case 1: When deleteDeltaStartTimestamp = deleteDeltaEndTimestamp. in this case only 1
        // delta file is present and deltaFileStamps is NULL
        // Case 2: When deleteDeltaStartTimestamp != deleteDeltaEndTimestamp. in this case more
        // than 1 delta files are present, then can blindly read deltaFilesStamps variable
        Arrays.stream(updateDetails).forEach(block -> {
          if (block.getDeleteDeltaStartTimestamp().equals(block.getDeleteDeltaEndTimestamp())) {
            totalDeltaFiles.removeIf(deltaFile -> deltaFile.getName().endsWith(block
                .getDeleteDeltaEndTimestamp() + CarbonCommonConstants.DELETE_DELTA_FILE_EXT));
          } else {
            block.getDeltaFileStamps().stream().forEach(fileName -> totalDeltaFiles
                .removeIf(deltaFile -> deltaFile.getName().endsWith(fileName +
                CarbonCommonConstants.DELETE_DELTA_FILE_EXT)));
          }
        });
        for (CarbonFile invalidFile: totalDeltaFiles) {
          totalSizeDeleted += invalidFile.getSize();
          filesToBeDeleted.add(invalidFile);
        }
      } else {
        for (LoadMetadataDetails segment : details) {
          // if this segment is valid then only we will go for delta file deletion.
          // if the segment is mark for delete or compacted then any way it will get deleted.
          if (segment.getSegmentStatus() == SegmentStatus.SUCCESS
                  || segment.getSegmentStatus() == SegmentStatus.LOAD_PARTIAL_SUCCESS) {
            // when there is no update operations done on table, then no need to go ahead. So
            // just check the update delta start timestamp and proceed if not empty
            if (!segment.getUpdateDeltaStartTimestamp().isEmpty()
                    || updatedSegments.contains(segment.getLoadName())) {
              // take the list of files from this segment.
              String segmentPath = CarbonTablePath.getSegmentPath(
                      table.getAbsoluteTableIdentifier().getTablePath(), segment.getLoadName());
              CarbonFile segDir =
                      FileFactory.getCarbonFile(segmentPath);
              CarbonFile[] allSegmentFiles = segDir.listFiles();
              SegmentUpdateDetails[] updateDetails = updateStatusManager.readLoadMetadata();
              for (SegmentUpdateDetails block : updateDetails) {
                CarbonFile[] invalidDeleteDeltaFiles;
                if (!block.getSegmentName().equalsIgnoreCase(segment.getLoadName())) {
                  continue;
                }
                invalidDeleteDeltaFiles = updateStatusManager.getDeleteDeltaInvalidFilesList(block,
                        false, allSegmentFiles, isInvalidFile);
                for (CarbonFile invalidFile : invalidDeleteDeltaFiles) {
                  totalSizeDeleted += invalidFile.getSize();
                  filesToBeDeleted.add(invalidFile);
                }
              }
            }
          }
        }
      }
  
      // delete the update table status files which are old.
      if (null != validUpdateStatusFile && !validUpdateStatusFile.isEmpty()) {
  
        final String updateStatusTimestamp = validUpdateStatusFile
                .substring(validUpdateStatusFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1);
  
        String tablePath = table.getAbsoluteTableIdentifier().getTablePath();
        CarbonFile metaFolder = FileFactory.getCarbonFile(
                CarbonTablePath.getMetadataPath(tablePath));
  
        CarbonFile[] invalidUpdateStatusFiles = metaFolder.listFiles(new CarbonFileFilter() {
          @Override
          public boolean accept(CarbonFile file) {
            if (file.getName().startsWith(CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME)) {
              // CHECK if this is valid or not.
              // we only send invalid ones to delete.
              return !file.getName().endsWith(updateStatusTimestamp);
            }
            return false;
          }
        });
  
        for (CarbonFile invalidFile : invalidUpdateStatusFiles) {
          totalSizeDeleted += invalidFile.getSize();
          filesToBeDeleted.add(invalidFile);
        }
      }
      if (!isDryRun) {
        for (CarbonFile invalidFile : filesToBeDeleted) {
          invalidFile.deleteFile();
        }
      }
      return totalSizeDeleted;
    }
  }
  
  