carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [25/49] incubator-carbondata git commit: Compaction lock should also be acquired during alter operation as alter and compaction on same table should not be allowed concurrently.
Date Fri, 07 Apr 2017 09:55:28 GMT
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
new file mode 100644
index 0000000..ee667c2
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
@@ -0,0 +1,1385 @@
+/*
+ * 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.merger;
+
+import java.io.File;
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.carbondata.common.logging.LogService;
+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.locks.CarbonLockFactory;
+import org.apache.carbondata.core.locks.ICarbonLock;
+import org.apache.carbondata.core.locks.LockUsage;
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.mutate.DeleteDeltaBlockDetails;
+import org.apache.carbondata.core.mutate.SegmentUpdateDetails;
+import org.apache.carbondata.core.reader.CarbonDeleteFilesDataReader;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+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.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+import org.apache.carbondata.core.writer.CarbonDeleteDeltaWriterImpl;
+import org.apache.carbondata.processing.model.CarbonLoadModel;
+
+/**
+ * utility class for load merging.
+ */
+public final class CarbonDataMergerUtil {
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(CarbonDataMergerUtil.class.getName());
+
+  private CarbonDataMergerUtil() {
+
+  }
+
+  /**
+   * Returns the size of all the carbondata files present in the segment.
+   * @param carbonFile
+   * @return
+   */
+  private static long getSizeOfFactFileInLoad(CarbonFile carbonFile) {
+    long factSize = 0;
+
+    // carbon data file case.
+    CarbonFile[] factFile = carbonFile.listFiles(new CarbonFileFilter() {
+
+      @Override public boolean accept(CarbonFile file) {
+        return CarbonTablePath.isCarbonDataFile(file.getName());
+      }
+    });
+
+    for (CarbonFile fact : factFile) {
+      factSize += fact.getSize();
+    }
+
+    return factSize;
+  }
+
+  /**
+   * To check whether the merge property is enabled or not.
+   *
+   * @return
+   */
+
+  public static boolean checkIfAutoLoadMergingRequired() {
+    // load merge is not supported as per new store format
+    // moving the load merge check in early to avoid unnecessary load listing causing IOException
+    // check whether carbons segment merging operation is enabled or not.
+    // default will be false.
+    String isLoadMergeEnabled = CarbonProperties.getInstance()
+        .getProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE,
+            CarbonCommonConstants.DEFAULT_ENABLE_AUTO_LOAD_MERGE);
+    if (isLoadMergeEnabled.equalsIgnoreCase("false")) {
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Form the Name of the New Merge Folder
+   *
+   * @param segmentsToBeMergedList
+   * @return
+   */
+  public static String getMergedLoadName(List<LoadMetadataDetails> segmentsToBeMergedList) {
+    String firstSegmentName = segmentsToBeMergedList.get(0).getLoadName();
+    if (firstSegmentName.contains(".")) {
+      String beforeDecimal = firstSegmentName.substring(0, firstSegmentName.indexOf("."));
+      String afterDecimal = firstSegmentName.substring(firstSegmentName.indexOf(".") + 1);
+      int fraction = Integer.parseInt(afterDecimal) + 1;
+      String mergedSegmentName = beforeDecimal + "." + fraction;
+      return CarbonCommonConstants.LOAD_FOLDER + mergedSegmentName;
+    } else {
+      String mergeName = firstSegmentName + "." + 1;
+      return CarbonCommonConstants.LOAD_FOLDER + mergeName;
+    }
+
+  }
+
+
+  /**
+   * Form the Name of the New Merge Folder
+   *
+   * @param segmentToBeMerged
+   * @return
+   */
+  public static String getMergedLoadName(final String segmentToBeMerged) {
+    String firstSegmentName = segmentToBeMerged;
+    if (firstSegmentName.contains(".")) {
+      String beforeDecimal = firstSegmentName.substring(0, firstSegmentName.indexOf("."));
+      String afterDecimal = firstSegmentName.substring(firstSegmentName.indexOf(".") + 1);
+      int fraction = Integer.parseInt(afterDecimal) + 1;
+      String mergedSegmentName = beforeDecimal + "." + fraction;
+      return mergedSegmentName;
+    } else {
+      String mergeName = firstSegmentName + "." + 1;
+      return mergeName;
+    }
+
+  }
+
+  /**
+   * Update Both Segment Update Status and Table Status for the case of IUD Delete
+   * delta compaction.
+   *
+   * @param loadsToMerge
+   * @param metaDataFilepath
+   * @param carbonLoadModel
+   * @return
+   */
+  public static boolean updateLoadMetadataIUDUpdateDeltaMergeStatus(
+      List<LoadMetadataDetails> loadsToMerge, String metaDataFilepath,
+      CarbonLoadModel carbonLoadModel) {
+
+    boolean status = false;
+    boolean updateLockStatus = false;
+    boolean tableLockStatus = false;
+
+    String timestamp = carbonLoadModel.getFactTimeStamp();
+
+    List<String> updatedDeltaFilesList =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    // This routine updateLoadMetadataIUDCompactionMergeStatus is suppose to update
+    // two files as it is only called during IUD_UPDDEL_DELTA_COMPACTION. Along with
+    // Table Status Metadata file (For Update Block Compaction) it has to update the
+    // Table Update Status Metadata File (For corresponding Delete Delta File).
+    // As the IUD_UPDDEL_DELTA_COMPACTION going to write in the same segment therefore in
+    // A) Table Update Status Metadata File (Block Level)
+    //      * For each blocks which is being compacted Mark 'Compacted' as the Status.
+    // B) Table Status Metadata file (Segment Level)
+    //      * loadStatus won't be changed to "compacted'
+    //      * UpdateDeltaStartTime and UpdateDeltaEndTime will be both set to current
+    //        timestamp (which is being passed from driver)
+    // First the Table Update Status Metadata File should be updated as we need to get
+    // the updated blocks for the segment from Table Status Metadata Update Delta Start and
+    // End Timestamp.
+
+    // Table Update Status Metadata Update.
+    AbsoluteTableIdentifier absoluteTableIdentifier =
+        carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    SegmentUpdateStatusManager segmentUpdateStatusManager =
+        new SegmentUpdateStatusManager(absoluteTableIdentifier);
+
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+
+    ICarbonLock updateLock = segmentUpdateStatusManager.getTableUpdateStatusLock();
+    ICarbonLock statusLock = segmentStatusManager.getTableStatusLock();
+
+    // Update the Compacted Blocks with Compacted Status.
+    try {
+      updatedDeltaFilesList = segmentUpdateStatusManager
+          .getUpdateDeltaFiles(loadsToMerge.get(0).getLoadName().toString());
+    } catch (Exception e) {
+      LOGGER.error("Error while getting the Update Delta Blocks.");
+      status = false;
+      return status;
+    }
+
+    if (updatedDeltaFilesList.size() > 0) {
+      try {
+        updateLockStatus = updateLock.lockWithRetries();
+        tableLockStatus = statusLock.lockWithRetries();
+
+        List<String> blockNames = new ArrayList<>(updatedDeltaFilesList.size());
+
+        for (String compactedBlocks : updatedDeltaFilesList) {
+          // Try to BlockName
+          String fullBlock = compactedBlocks;
+          int endIndex = fullBlock.lastIndexOf(File.separator);
+          String blkNoExt = fullBlock.substring(endIndex + 1, fullBlock.lastIndexOf("-"));
+          blockNames.add(blkNoExt);
+        }
+
+        if (updateLockStatus && tableLockStatus) {
+
+          SegmentUpdateDetails[] updateLists = segmentUpdateStatusManager
+              .readLoadMetadata();
+
+          for (String compactedBlocks : blockNames) {
+            // Check is the compactedBlocks name matches with oldDetails
+            for (int i = 0; i < updateLists.length; i++) {
+              if (updateLists[i].getBlockName().equalsIgnoreCase(compactedBlocks)
+                  && !CarbonCommonConstants.COMPACTED.equalsIgnoreCase(updateLists[i].getStatus())
+                  && !CarbonCommonConstants.MARKED_FOR_DELETE
+                  .equalsIgnoreCase(updateLists[i].getStatus())) {
+                updateLists[i].setStatus(CarbonCommonConstants.COMPACTED);
+              }
+            }
+          }
+
+          LoadMetadataDetails[] loadDetails =
+              segmentStatusManager.readLoadMetadata(metaDataFilepath);
+
+          for (LoadMetadataDetails loadDetail : loadDetails) {
+            if (loadsToMerge.contains(loadDetail)) {
+              loadDetail.setUpdateDeltaStartTimestamp(timestamp);
+              loadDetail.setUpdateDeltaEndTimestamp(timestamp);
+              if (loadDetail.getLoadName().equalsIgnoreCase("0")) {
+                loadDetail
+                    .setUpdateStatusFileName(CarbonUpdateUtil.getUpdateStatusFileName(timestamp));
+              }
+            }
+          }
+
+          try {
+            segmentUpdateStatusManager
+                .writeLoadDetailsIntoFile(Arrays.asList(updateLists), timestamp);
+            segmentStatusManager
+                .writeLoadDetailsIntoFile(carbonTablePath.getTableStatusFilePath(), loadDetails);
+            status = true;
+          } catch (IOException e) {
+            LOGGER.error(
+                "Error while writing metadata. The metadata file path is " + carbonTablePath
+                    .getMetadataDirectoryPath());
+            status = false;
+          }
+        } else {
+          LOGGER.error("Not able to acquire the lock.");
+          status = false;
+        }
+      } catch (Exception e) {
+        LOGGER.error("Error while updating metadata. The metadata file path is " + carbonTablePath
+            .getMetadataDirectoryPath());
+        status = false;
+
+      } finally {
+        if (updateLockStatus) {
+          if (updateLock.unlock()) {
+            LOGGER.info("Unlock the segment update lock successfully.");
+          } else {
+            LOGGER.error("Not able to unlock the segment update lock.");
+          }
+        }
+        if (tableLockStatus) {
+          if (statusLock.unlock()) {
+            LOGGER.info("Unlock the table status lock successfully.");
+          } else {
+            LOGGER.error("Not able to unlock the table status lock.");
+          }
+        }
+      }
+    }
+    return status;
+  }
+
+  /**
+   * method to update table status in case of IUD Update Delta Compaction.
+   * @param loadsToMerge
+   * @param metaDataFilepath
+   * @param MergedLoadName
+   * @param carbonLoadModel
+   * @param compactionType
+   * @return
+   */
+  public static boolean updateLoadMetadataWithMergeStatus(List<LoadMetadataDetails> loadsToMerge,
+      String metaDataFilepath, String MergedLoadName, CarbonLoadModel carbonLoadModel,
+      long mergeLoadStartTime, CompactionType compactionType) {
+
+    boolean tableStatusUpdationStatus = false;
+    AbsoluteTableIdentifier absoluteTableIdentifier =
+        carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+
+    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
+
+    try {
+      if (carbonLock.lockWithRetries()) {
+        LOGGER.info("Acquired lock for the table " + carbonLoadModel.getDatabaseName() + "."
+            + carbonLoadModel.getTableName() + " for table status updation ");
+
+        CarbonTablePath carbonTablePath = CarbonStorePath
+            .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+                absoluteTableIdentifier.getCarbonTableIdentifier());
+
+        String statusFilePath = carbonTablePath.getTableStatusFilePath();
+
+        LoadMetadataDetails[] loadDetails = SegmentStatusManager.readLoadMetadata(metaDataFilepath);
+
+        String mergedLoadNumber = MergedLoadName.substring(
+            MergedLoadName.lastIndexOf(CarbonCommonConstants.LOAD_FOLDER)
+                + CarbonCommonConstants.LOAD_FOLDER.length(), MergedLoadName.length());
+
+        long modificationOrDeletionTimeStamp = CarbonUpdateUtil.readCurrentTime();
+        for (LoadMetadataDetails loadDetail : loadDetails) {
+          // check if this segment is merged.
+          if (loadsToMerge.contains(loadDetail)) {
+            // if the compacted load is deleted after the start of the compaction process,
+            // then need to discard the compaction process and treat it as failed compaction.
+            if (loadDetail.getLoadStatus()
+                .equalsIgnoreCase(CarbonCommonConstants.MARKED_FOR_DELETE)) {
+              LOGGER.error("Compaction is aborted as the segment " + loadDetail.getLoadName()
+                  + " is deleted after the compaction is started.");
+              return false;
+            }
+            loadDetail.setLoadStatus(CarbonCommonConstants.COMPACTED);
+            loadDetail.setModificationOrdeletionTimesStamp(modificationOrDeletionTimeStamp);
+            loadDetail.setMergedLoadName(mergedLoadNumber);
+          }
+        }
+
+        // create entry for merged one.
+        LoadMetadataDetails loadMetadataDetails = new LoadMetadataDetails();
+        loadMetadataDetails.setPartitionCount(carbonLoadModel.getPartitionId());
+        loadMetadataDetails.setLoadStatus(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS);
+        long loadEnddate = CarbonUpdateUtil.readCurrentTime();
+        loadMetadataDetails.setLoadEndTime(loadEnddate);
+        loadMetadataDetails.setLoadName(mergedLoadNumber);
+        loadMetadataDetails.setLoadStartTime(mergeLoadStartTime);
+        loadMetadataDetails.setPartitionCount("0");
+        // if this is a major compaction then set the segment as major compaction.
+        if (compactionType == CompactionType.MAJOR_COMPACTION) {
+          loadMetadataDetails.setMajorCompacted("true");
+        }
+
+        List<LoadMetadataDetails> updatedDetailsList = new ArrayList<>(Arrays.asList(loadDetails));
+
+        // put the merged folder entry
+        updatedDetailsList.add(loadMetadataDetails);
+
+        try {
+          SegmentStatusManager.writeLoadDetailsIntoFile(statusFilePath,
+              updatedDetailsList.toArray(new LoadMetadataDetails[updatedDetailsList.size()]));
+          tableStatusUpdationStatus = true;
+        } catch (IOException e) {
+          LOGGER.error("Error while writing metadata");
+        }
+      } else {
+        LOGGER.error(
+            "Could not able to obtain lock for table" + carbonLoadModel.getDatabaseName() + "."
+                + carbonLoadModel.getTableName() + "for table status updation");
+      }
+    } finally {
+      if (carbonLock.unlock()) {
+        LOGGER.info("Table unlocked successfully after table status updation" + carbonLoadModel
+            .getDatabaseName() + "." + carbonLoadModel.getTableName());
+      } else {
+        LOGGER.error(
+            "Unable to unlock Table lock for table" + carbonLoadModel.getDatabaseName() + "."
+                + carbonLoadModel.getTableName() + " during table status updation");
+      }
+    }
+    return tableStatusUpdationStatus;
+  }
+
+  /**
+   * To identify which all segments can be merged.
+   *
+   * @param storeLocation
+   * @param carbonLoadModel
+   * @param compactionSize
+   * @return
+   */
+  public static List<LoadMetadataDetails> identifySegmentsToBeMerged(String storeLocation,
+      CarbonLoadModel carbonLoadModel, long compactionSize,
+      List<LoadMetadataDetails> segments, CompactionType compactionType) {
+
+    List<LoadMetadataDetails> sortedSegments = new ArrayList<LoadMetadataDetails>(segments);
+
+    sortSegments(sortedSegments);
+
+    // Check for segments which are qualified for IUD compaction.
+    if (compactionType.equals(CompactionType.IUD_UPDDEL_DELTA_COMPACTION)) {
+
+      List<LoadMetadataDetails> listOfSegmentsToBeMerged =
+          identifySegmentsToBeMergedBasedOnIUD(sortedSegments, carbonLoadModel);
+
+      return listOfSegmentsToBeMerged;
+    }
+
+    // check preserve property and preserve the configured number of latest loads.
+
+    List<LoadMetadataDetails> listOfSegmentsAfterPreserve =
+        checkPreserveSegmentsPropertyReturnRemaining(sortedSegments);
+
+    // filter the segments if the compaction based on days is configured.
+
+    List<LoadMetadataDetails> listOfSegmentsLoadedInSameDateInterval =
+        identifySegmentsToBeMergedBasedOnLoadedDate(listOfSegmentsAfterPreserve);
+    List<LoadMetadataDetails> listOfSegmentsToBeMerged;
+    // identify the segments to merge based on the Size of the segments across partition.
+    if (compactionType.equals(CompactionType.MAJOR_COMPACTION)) {
+
+      listOfSegmentsToBeMerged = identifySegmentsToBeMergedBasedOnSize(compactionSize,
+          listOfSegmentsLoadedInSameDateInterval, carbonLoadModel, storeLocation);
+    } else {
+
+      listOfSegmentsToBeMerged =
+          identifySegmentsToBeMergedBasedOnSegCount(listOfSegmentsLoadedInSameDateInterval);
+    }
+
+    return listOfSegmentsToBeMerged;
+  }
+
+  /**
+   * Sorting of the segments.
+   * @param segments
+   */
+  public static void sortSegments(List segments) {
+    // sort the segment details.
+    Collections.sort(segments, new Comparator<LoadMetadataDetails>() {
+      @Override public int compare(LoadMetadataDetails seg1, LoadMetadataDetails seg2) {
+        double seg1Id = Double.parseDouble(seg1.getLoadName());
+        double seg2Id = Double.parseDouble(seg2.getLoadName());
+        if (seg1Id - seg2Id < 0) {
+          return -1;
+        }
+        if (seg1Id - seg2Id > 0) {
+          return 1;
+        }
+        return 0;
+      }
+    });
+  }
+
+  /**
+   * This method will return the list of loads which are loaded at the same interval.
+   * This property is configurable.
+   *
+   * @param listOfSegmentsBelowThresholdSize
+   * @return
+   */
+  private static List<LoadMetadataDetails> identifySegmentsToBeMergedBasedOnLoadedDate(
+      List<LoadMetadataDetails> listOfSegmentsBelowThresholdSize) {
+
+    List<LoadMetadataDetails> loadsOfSameDate =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    long numberOfDaysAllowedToMerge = 0;
+    try {
+      numberOfDaysAllowedToMerge = Long.parseLong(CarbonProperties.getInstance()
+          .getProperty(CarbonCommonConstants.DAYS_ALLOWED_TO_COMPACT,
+              CarbonCommonConstants.DEFAULT_DAYS_ALLOWED_TO_COMPACT));
+      if (numberOfDaysAllowedToMerge < 0 || numberOfDaysAllowedToMerge > 100) {
+        LOGGER.error(
+            "The specified value for property " + CarbonCommonConstants.DAYS_ALLOWED_TO_COMPACT
+                + " is incorrect."
+                + " Correct value should be in range of 0 -100. Taking the default value.");
+        numberOfDaysAllowedToMerge =
+            Long.parseLong(CarbonCommonConstants.DEFAULT_DAYS_ALLOWED_TO_COMPACT);
+      }
+
+    } catch (NumberFormatException e) {
+      numberOfDaysAllowedToMerge =
+          Long.parseLong(CarbonCommonConstants.DEFAULT_DAYS_ALLOWED_TO_COMPACT);
+    }
+    // if true then process loads according to the load date.
+    if (numberOfDaysAllowedToMerge > 0) {
+
+      // filter loads based on the loaded date
+      boolean first = true;
+      Date segDate1 = null;
+      SimpleDateFormat sdf = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP);
+      for (LoadMetadataDetails segment : listOfSegmentsBelowThresholdSize) {
+
+        if (first) {
+          segDate1 = initializeFirstSegment(loadsOfSameDate, segment, sdf);
+          first = false;
+          continue;
+        }
+        long segmentDate = segment.getLoadStartTime();
+        Date segDate2 = null;
+        try {
+          segDate2 = sdf.parse(sdf.format(segmentDate));
+        } catch (ParseException e) {
+          LOGGER.error("Error while parsing segment start time" + e.getMessage());
+        }
+
+        if (isTwoDatesPresentInRequiredRange(segDate1, segDate2, numberOfDaysAllowedToMerge)) {
+          loadsOfSameDate.add(segment);
+        }
+        // if the load is beyond merged date.
+        // then reset everything and continue search for loads.
+        else if (loadsOfSameDate.size() < 2) {
+          loadsOfSameDate.clear();
+          // need to add the next segment as first and  to check further
+          segDate1 = initializeFirstSegment(loadsOfSameDate, segment, sdf);
+        } else { // case where a load is beyond merge date and there is at least 2 loads to merge.
+          break;
+        }
+      }
+    } else {
+      return listOfSegmentsBelowThresholdSize;
+    }
+
+    return loadsOfSameDate;
+  }
+
+  /**
+   * @param loadsOfSameDate
+   * @param segment
+   * @return
+   */
+  private static Date initializeFirstSegment(List<LoadMetadataDetails> loadsOfSameDate,
+      LoadMetadataDetails segment, SimpleDateFormat sdf) {
+    long baselineLoadStartTime = segment.getLoadStartTime();
+    Date segDate1 = null;
+    try {
+      segDate1 = sdf.parse(sdf.format(baselineLoadStartTime));
+    } catch (ParseException e) {
+      LOGGER.error("Error while parsing segment start time" + e.getMessage());
+    }
+    loadsOfSameDate.add(segment);
+    return segDate1;
+  }
+
+  /**
+   * Method to check if the load dates are complied to the configured dates.
+   *
+   * @param segDate1
+   * @param segDate2
+   * @return
+   */
+  private static boolean isTwoDatesPresentInRequiredRange(Date segDate1, Date segDate2,
+      long numberOfDaysAllowedToMerge) {
+    if (segDate1 == null || segDate2 == null) {
+      return false;
+    }
+    // take 1 st date add the configured days .
+    Calendar cal1 = Calendar.getInstance();
+    cal1.set(segDate1.getYear(), segDate1.getMonth(), segDate1.getDate());
+    Calendar cal2 = Calendar.getInstance();
+    cal2.set(segDate2.getYear(), segDate2.getMonth(), segDate2.getDate());
+
+    long diff = cal2.getTimeInMillis() - cal1.getTimeInMillis();
+
+    if ((diff / (24 * 60 * 60 * 1000)) < numberOfDaysAllowedToMerge) {
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Identify the segments to be merged based on the Size in case of Major compaction.
+   *
+   * @param compactionSize
+   * @param listOfSegmentsAfterPreserve
+   * @param carbonLoadModel
+   * @param storeLocation
+   * @return
+   */
+  private static List<LoadMetadataDetails> identifySegmentsToBeMergedBasedOnSize(
+      long compactionSize, List<LoadMetadataDetails> listOfSegmentsAfterPreserve,
+      CarbonLoadModel carbonLoadModel, String storeLocation) {
+
+    List<LoadMetadataDetails> segmentsToBeMerged =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    CarbonTableIdentifier tableIdentifier =
+        carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable().getCarbonTableIdentifier();
+
+
+    // total length
+    long totalLength = 0;
+
+    // check size of each segment , sum it up across partitions
+    for (LoadMetadataDetails segment : listOfSegmentsAfterPreserve) {
+
+      String segId = segment.getLoadName();
+      // variable to store one  segment size across partition.
+      long sizeOfOneSegmentAcrossPartition =
+          getSizeOfSegment(storeLocation, tableIdentifier, segId);
+
+      // if size of a segment is greater than the Major compaction size. then ignore it.
+      if (sizeOfOneSegmentAcrossPartition > (compactionSize * 1024 * 1024)) {
+        // if already 2 segments have been found for merging then stop scan here and merge.
+        if (segmentsToBeMerged.size() > 1) {
+          break;
+        } else { // if only one segment is found then remove the earlier one in list.
+          // reset the total length to 0.
+          segmentsToBeMerged = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+          totalLength = 0;
+          continue;
+        }
+      }
+
+      totalLength += sizeOfOneSegmentAcrossPartition;
+
+      // in case of major compaction the size doesnt matter. all the segments will be merged.
+      if (totalLength < (compactionSize * 1024 * 1024)) {
+        segmentsToBeMerged.add(segment);
+      } else { // if already 2 segments have been found for merging then stop scan here and merge.
+        if (segmentsToBeMerged.size() > 1) {
+          break;
+        } else { // if only one segment is found then remove the earlier one in list and put this.
+          // reset the total length to the current identified segment.
+          segmentsToBeMerged = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+          segmentsToBeMerged.add(segment);
+          totalLength = sizeOfOneSegmentAcrossPartition;
+        }
+      }
+
+    }
+
+    return segmentsToBeMerged;
+  }
+
+  /**
+   * For calculating the size of the specified segment
+   * @param storeLocation
+   * @param tableIdentifier
+   * @param segId
+   * @return
+   */
+  private static long getSizeOfSegment(String storeLocation,
+      CarbonTableIdentifier tableIdentifier, String segId) {
+    String loadPath = getStoreLocation(storeLocation, tableIdentifier, segId);
+    CarbonFile segmentFolder =
+        FileFactory.getCarbonFile(loadPath, FileFactory.getFileType(loadPath));
+    return getSizeOfFactFileInLoad(segmentFolder);
+  }
+
+  /**
+   * This method will get the store location for the given path, segemnt id and partition id
+   *
+   * @param storePath
+   * @param carbonTableIdentifier
+   * @param segmentId
+   * @return
+   */
+  private static String getStoreLocation(String storePath,
+      CarbonTableIdentifier carbonTableIdentifier, String segmentId) {
+    CarbonTablePath carbonTablePath =
+        CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier);
+    return carbonTablePath.getCarbonDataDirectoryPath("0", segmentId);
+  }
+
+
+  /**
+   * Identify the segments to be merged based on the segment count
+   *
+   * @param listOfSegmentsAfterPreserve
+   * @return
+   */
+  private static List<LoadMetadataDetails> identifySegmentsToBeMergedBasedOnSegCount(
+      List<LoadMetadataDetails> listOfSegmentsAfterPreserve) {
+
+    List<LoadMetadataDetails> mergedSegments =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    List<LoadMetadataDetails> unMergedSegments =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    int[] noOfSegmentLevelsCount =
+        CarbonProperties.getInstance().getCompactionSegmentLevelCount();
+
+    int level1Size = 0;
+    int level2Size = 0;
+    int size = noOfSegmentLevelsCount.length;
+
+    if (size >= 2) {
+      level1Size = noOfSegmentLevelsCount[0];
+      level2Size = noOfSegmentLevelsCount[1];
+    } else if (size == 1) {
+      level1Size = noOfSegmentLevelsCount[0];
+    }
+
+    int unMergeCounter = 0;
+    int mergeCounter = 0;
+
+    // check size of each segment , sum it up across partitions
+    for (LoadMetadataDetails segment : listOfSegmentsAfterPreserve) {
+
+      String segName = segment.getLoadName();
+
+      // if a segment is already merged 2 levels then it s name will become .2
+      // need to exclude those segments from minor compaction.
+      // if a segment is major compacted then should not be considered for minor.
+      if (segName.endsWith(CarbonCommonConstants.LEVEL2_COMPACTION_INDEX) || (
+          segment.isMajorCompacted() != null && segment.isMajorCompacted()
+              .equalsIgnoreCase("true"))) {
+        continue;
+      }
+
+      // check if the segment is merged or not
+
+      if (!isMergedSegment(segName)) {
+        //if it is an unmerged segment then increment counter
+        unMergeCounter++;
+        unMergedSegments.add(segment);
+        if (unMergeCounter == (level1Size)) {
+          return unMergedSegments;
+        }
+      } else {
+        mergeCounter++;
+        mergedSegments.add(segment);
+        if (mergeCounter == (level2Size)) {
+          return mergedSegments;
+        }
+      }
+    }
+    return new ArrayList<>(0);
+  }
+
+  /**
+   * To check if the segment is merged or not.
+   * @param segName
+   * @return
+   */
+  private static boolean isMergedSegment(String segName) {
+    if (segName.contains(".")) {
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * checks number of loads to be preserved and returns remaining valid segments
+   *
+   * @param segments
+   * @return
+   */
+  private static List<LoadMetadataDetails> checkPreserveSegmentsPropertyReturnRemaining(
+      List<LoadMetadataDetails> segments) {
+    // check whether the preserving of the segments from merging is enabled or not.
+    // get the number of loads to be preserved.
+    int numberOfSegmentsToBePreserved =
+        CarbonProperties.getInstance().getNumberOfSegmentsToBePreserved();
+    // get the number of valid segments and retain the latest loads from merging.
+    return CarbonDataMergerUtil
+        .getValidLoadDetailsWithRetaining(segments, numberOfSegmentsToBePreserved);
+  }
+
+  /**
+   * Retain the number of segments which are to be preserved and return the remaining list of
+   * segments.
+   *
+   * @param loadMetadataDetails
+   * @param numberOfSegToBeRetained
+   * @return
+   */
+  private static List<LoadMetadataDetails> getValidLoadDetailsWithRetaining(
+      List<LoadMetadataDetails> loadMetadataDetails, int numberOfSegToBeRetained) {
+
+    List<LoadMetadataDetails> validList =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    for (LoadMetadataDetails segment : loadMetadataDetails) {
+      if (isSegmentValid(segment)) {
+        validList.add(segment);
+      }
+    }
+
+    // check if valid list is big enough for removing the number of seg to be retained.
+    // last element
+    int removingIndex = validList.size() - 1;
+
+    for (int i = validList.size(); i > 0; i--) {
+      if (numberOfSegToBeRetained == 0) {
+        break;
+      }
+      // remove last segment
+      validList.remove(removingIndex--);
+      numberOfSegToBeRetained--;
+    }
+    return validList;
+
+  }
+
+  /**
+   * This will give the compaction sizes configured based on compaction type.
+   *
+   * @param compactionType
+   * @return
+   */
+  public static long getCompactionSize(CompactionType compactionType) {
+
+    long compactionSize = 0;
+    switch (compactionType) {
+      case MAJOR_COMPACTION:
+        compactionSize = CarbonProperties.getInstance().getMajorCompactionSize();
+        break;
+      default: // this case can not come.
+    }
+    return compactionSize;
+  }
+
+  /**
+   * For getting the comma separated valid segments for merging.
+   *
+   * @param loadMetadataDetails
+   * @return
+   */
+  public static String getValidSegments(List<LoadMetadataDetails> loadMetadataDetails) {
+    StringBuilder builder = new StringBuilder();
+    for (LoadMetadataDetails segment : loadMetadataDetails) {
+      //check if this load is an already merged load.
+      if (null != segment.getMergedLoadName()) {
+        builder.append(segment.getMergedLoadName()).append(",");
+      } else {
+        builder.append(segment.getLoadName()).append(",");
+      }
+    }
+    builder.deleteCharAt(builder.length() - 1);
+    return builder.toString();
+  }
+
+  /**
+   * This method returns the valid segments attached to the table Identifier.
+   *
+   * @param absoluteTableIdentifier
+   * @return
+   */
+  public static List<String> getValidSegmentList(AbsoluteTableIdentifier absoluteTableIdentifier)
+          throws IOException {
+
+    SegmentStatusManager.ValidAndInvalidSegmentsInfo validAndInvalidSegments = null;
+    try {
+      validAndInvalidSegments =
+              new SegmentStatusManager(absoluteTableIdentifier).getValidAndInvalidSegments();
+    } catch (IOException e) {
+      LOGGER.error("Error while getting valid segment list for a table identifier");
+      throw new IOException();
+    }
+    return validAndInvalidSegments.getValidSegments();
+  }
+
+
+  /**
+   * Removing the already merged segments from list.
+   */
+  public static List<LoadMetadataDetails> filterOutNewlyAddedSegments(
+      List<LoadMetadataDetails> segments,
+      LoadMetadataDetails lastSeg) {
+
+    // take complete list of segments.
+    List<LoadMetadataDetails> list = new ArrayList<>(segments);
+    // sort list
+    CarbonDataMergerUtil.sortSegments(list);
+
+    // first filter out newly added segments.
+    return list.subList(0, list.indexOf(lastSeg) + 1);
+
+  }
+
+  /**
+   * method to identify the segments qualified for merging in case of IUD Compaction.
+   *
+   * @param carbonLoadModel
+   * @param compactionType
+   * @return
+   */
+  private static List<LoadMetadataDetails> identifySegmentsToBeMergedBasedOnIUD(
+      List<LoadMetadataDetails> segments, CarbonLoadModel carbonLoadModel) {
+
+    List<LoadMetadataDetails> validSegments = new ArrayList<>(segments.size());
+
+    AbsoluteTableIdentifier absoluteTableIdentifier =
+        carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
+
+    int numberUpdateDeltaFilesThreshold =
+        CarbonProperties.getInstance().getNoUpdateDeltaFilesThresholdForIUDCompaction();
+    for (LoadMetadataDetails seg : segments) {
+      if ((isSegmentValid(seg)) && checkUpdateDeltaFilesInSeg(seg.getLoadName(),
+          absoluteTableIdentifier, carbonLoadModel.getSegmentUpdateStatusManager(),
+          numberUpdateDeltaFilesThreshold)) {
+        validSegments.add(seg);
+      }
+    }
+    return validSegments;
+  }
+
+  private static boolean isSegmentValid(LoadMetadataDetails seg) {
+    return seg.getLoadStatus().equalsIgnoreCase(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS)
+            || seg.getLoadStatus()
+            .equalsIgnoreCase(CarbonCommonConstants.STORE_LOADSTATUS_PARTIAL_SUCCESS) || seg
+            .getLoadStatus().equalsIgnoreCase(CarbonCommonConstants.MARKED_FOR_UPDATE);
+  }
+
+  /**
+   * method gets the segments list which get qualified for IUD compaction.
+   * @param Segments
+   * @param absoluteTableIdentifier
+   * @param compactionTypeIUD
+   * @return
+   */
+  public static List<String> getSegListIUDCompactionQualified(List<String> Segments,
+      AbsoluteTableIdentifier absoluteTableIdentifier,
+      SegmentUpdateStatusManager segmentUpdateStatusManager, CompactionType compactionTypeIUD) {
+
+    List<String> validSegments = new ArrayList<>();
+
+    if (compactionTypeIUD.equals(CompactionType.IUD_DELETE_DELTA_COMPACTION)) {
+      int numberDeleteDeltaFilesThreshold =
+          CarbonProperties.getInstance().getNoDeleteDeltaFilesThresholdForIUDCompaction();
+      List<String> deleteSegments = new ArrayList<>();
+      for (String seg : Segments) {
+        if (checkDeleteDeltaFilesInSeg(seg, segmentUpdateStatusManager,
+            numberDeleteDeltaFilesThreshold)) {
+          deleteSegments.add(seg);
+        }
+      }
+      if (deleteSegments.size() > 0) {
+        // This Code Block Append the Segname along with the Blocks selected for Merge instead of
+        // only taking the segment name. This will help to parallelize better for each block
+        // in case of Delete Horizontal Compaction.
+        for (String segName : deleteSegments) {
+          List<String> tempSegments = getDeleteDeltaFilesInSeg(segName, segmentUpdateStatusManager,
+              numberDeleteDeltaFilesThreshold);
+          if (tempSegments != null) {
+            for (String tempSeg : tempSegments) {
+              validSegments.add(tempSeg);
+            }
+          }
+        }
+      }
+    } else if (compactionTypeIUD.equals(CompactionType.IUD_UPDDEL_DELTA_COMPACTION)) {
+      int numberUpdateDeltaFilesThreshold =
+          CarbonProperties.getInstance().getNoUpdateDeltaFilesThresholdForIUDCompaction();
+      for (String seg : Segments) {
+        if (checkUpdateDeltaFilesInSeg(seg, absoluteTableIdentifier, segmentUpdateStatusManager,
+            numberUpdateDeltaFilesThreshold)) {
+          validSegments.add(seg);
+        }
+      }
+    }
+    return validSegments;
+  }
+
+  /**
+   * Check if the blockname of the segment belongs to the Valid Update Delta List or not.
+   * @param seg
+   * @param blkName
+   * @param segmentUpdateStatusManager
+   * @return
+   */
+  public static Boolean checkUpdateDeltaMatchBlock(final String seg, final String blkName,
+      SegmentUpdateStatusManager segmentUpdateStatusManager) {
+
+    List<String> list = segmentUpdateStatusManager.getUpdateDeltaFiles(seg);
+
+    String fullBlock = blkName;
+    String[] FileParts = fullBlock.split(CarbonCommonConstants.FILE_SEPARATOR);
+    String blockName = FileParts[FileParts.length - 1];
+
+    for (String str : list) {
+      if (str.contains(blockName)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * This method traverses Update Delta Files inside the seg and return true
+   * if UpdateDelta Files are more than IUD Compaction threshold.
+   *
+   * @param seg
+   * @param absoluteTableIdentifier
+   * @param segmentUpdateStatusManager
+   * @param numberDeltaFilesThreshold
+   * @return
+   */
+  public static Boolean checkUpdateDeltaFilesInSeg(String seg,
+      AbsoluteTableIdentifier absoluteTableIdentifier,
+      SegmentUpdateStatusManager segmentUpdateStatusManager, int numberDeltaFilesThreshold) {
+
+    CarbonFile[] updateDeltaFiles = null;
+    Set<String> uniqueBlocks = new HashSet<String>();
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    String segmentPath = carbonTablePath.getCarbonDataDirectoryPath("0", seg);
+    CarbonFile segDir =
+        FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+    CarbonFile[] allSegmentFiles = segDir.listFiles();
+
+    updateDeltaFiles = segmentUpdateStatusManager
+        .getUpdateDeltaFilesForSegment(seg, true, CarbonCommonConstants.UPDATE_DELTA_FILE_EXT,
+            false, allSegmentFiles);
+
+    if (updateDeltaFiles == null) {
+      return false;
+    }
+
+    // The update Delta files may have Spill over blocks. Will consider multiple spill over
+    // blocks as one. Currently updateDeltaFiles array contains Update Delta Block name which
+    // lies within UpdateDelta Start TimeStamp and End TimeStamp. In order to eliminate
+    // Spill Over Blocks will choose files with unique taskID.
+    for (CarbonFile blocks : updateDeltaFiles) {
+      // Get Task ID and the Timestamp from the Block name for e.g.
+      // part-0-3-1481084721319.carbondata => "3-1481084721319"
+      String task = CarbonTablePath.DataFileUtil.getTaskNo(blocks.getName());
+      String timestamp =
+          CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(blocks.getName());
+      String taskAndTimeStamp = task + "-" + timestamp;
+      uniqueBlocks.add(taskAndTimeStamp);
+    }
+    if (uniqueBlocks.size() > numberDeltaFilesThreshold) {
+      return true;
+    } else {
+      return false;
+    }
+  }
+
+  /**
+   * Check is the segment passed qualifies for IUD delete delta compaction or not i.e.
+   * if the number of delete delta files present in the segment is more than
+   * numberDeltaFilesThreshold.
+   *
+   * @param seg
+   * @param segmentUpdateStatusManager
+   * @param numberDeltaFilesThreshold
+   * @return
+   */
+  private static boolean checkDeleteDeltaFilesInSeg(String seg,
+      SegmentUpdateStatusManager segmentUpdateStatusManager, int numberDeltaFilesThreshold) {
+
+    Set<String> uniqueBlocks = new HashSet<String>();
+    List<String> blockNameList = segmentUpdateStatusManager.getBlockNameFromSegment(seg);
+
+    for (final String blockName : blockNameList) {
+
+      CarbonFile[] deleteDeltaFiles =
+          segmentUpdateStatusManager.getDeleteDeltaFilesList(seg, blockName);
+
+      // The Delete Delta files may have Spill over blocks. Will consider multiple spill over
+      // blocks as one. Currently DeleteDeltaFiles array contains Delete Delta Block name which
+      // lies within Delete Delta Start TimeStamp and End TimeStamp. In order to eliminate
+      // Spill Over Blocks will choose files with unique taskID.
+      for (CarbonFile blocks : deleteDeltaFiles) {
+        // Get Task ID and the Timestamp from the Block name for e.g.
+        // part-0-3-1481084721319.carbondata => "3-1481084721319"
+        String task = CarbonTablePath.DataFileUtil.getTaskNo(blocks.getName());
+        String timestamp =
+            CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(blocks.getName());
+        String taskAndTimeStamp = task + "-" + timestamp;
+        uniqueBlocks.add(taskAndTimeStamp);
+      }
+
+      if (uniqueBlocks.size() > numberDeltaFilesThreshold) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Check is the segment passed qualifies for IUD delete delta compaction or not i.e.
+   * if the number of delete delta files present in the segment is more than
+   * numberDeltaFilesThreshold.
+   * @param seg
+   * @param segmentUpdateStatusManager
+   * @param numberDeltaFilesThreshold
+   * @return
+   */
+
+  private static List<String> getDeleteDeltaFilesInSeg(String seg,
+      SegmentUpdateStatusManager segmentUpdateStatusManager, int numberDeltaFilesThreshold) {
+
+    List<String> blockLists = new ArrayList<>();
+    List<String> blockNameList = segmentUpdateStatusManager.getBlockNameFromSegment(seg);
+
+    for (final String blockName : blockNameList) {
+
+      CarbonFile[] deleteDeltaFiles =
+          segmentUpdateStatusManager.getDeleteDeltaFilesList(seg, blockName);
+
+      if (deleteDeltaFiles.length > numberDeltaFilesThreshold) {
+        blockLists.add(seg + "/" + blockName);
+      }
+    }
+    return blockLists;
+  }
+
+  /**
+   * Returns true is horizontal compaction is enabled.
+   * @return
+   */
+  public static boolean isHorizontalCompactionEnabled() {
+    if ((CarbonProperties.getInstance()
+        .getProperty(CarbonCommonConstants.isHorizontalCompactionEnabled,
+            CarbonCommonConstants.defaultIsHorizontalCompactionEnabled)).equalsIgnoreCase("true")) {
+      return true;
+    } else {
+      return false;
+    }
+  }
+
+  /**
+   * method to compact Delete Delta files in case of IUD Compaction.
+   *
+   * @param seg
+   * @param blockName
+   * @param absoluteTableIdentifier
+   * @param segmentUpdateDetails
+   * @param timestamp
+   * @return
+   * @throws IOException
+   */
+  public static List<CarbonDataMergerUtilResult> compactBlockDeleteDeltaFiles(String seg,
+      String blockName, AbsoluteTableIdentifier absoluteTableIdentifier,
+      SegmentUpdateDetails[] segmentUpdateDetails, Long timestamp) throws IOException {
+
+    SegmentUpdateStatusManager segmentUpdateStatusManager =
+        new SegmentUpdateStatusManager(absoluteTableIdentifier);
+
+    List<CarbonDataMergerUtilResult> resultList = new ArrayList<CarbonDataMergerUtilResult>(1);
+
+    // set the update status.
+    segmentUpdateStatusManager.setUpdateStatusDetails(segmentUpdateDetails);
+
+    CarbonFile[] deleteDeltaFiles =
+        segmentUpdateStatusManager.getDeleteDeltaFilesList(seg, blockName);
+
+    String destFileName =
+        blockName + "-" + timestamp.toString() + CarbonCommonConstants.DELETE_DELTA_FILE_EXT;
+    String fullBlockFilePath = deleteDeltaFiles[0].getParentFile().getCanonicalPath()
+        + CarbonCommonConstants.FILE_SEPARATOR + destFileName;
+
+    List<String> deleteFilePathList = new ArrayList<String>();
+    for (CarbonFile cFile : deleteDeltaFiles) {
+      deleteFilePathList.add(cFile.getCanonicalPath());
+    }
+
+    CarbonDataMergerUtilResult blockDetails = new CarbonDataMergerUtilResult();
+    blockDetails.setBlockName(blockName);
+    blockDetails.setSegmentName(seg);
+    blockDetails.setDeleteDeltaStartTimestamp(timestamp.toString());
+    blockDetails.setDeleteDeltaEndTimestamp(timestamp.toString());
+
+    try {
+      if (startCompactionDeleteDeltaFiles(deleteFilePathList, blockName, fullBlockFilePath)) {
+        blockDetails.setCompactionStatus(true);
+      } else {
+        blockDetails.setCompactionStatus(false);
+      }
+      resultList.add(blockDetails);
+    } catch (IOException e) {
+      LOGGER.error("Compaction of Delete Delta Files failed. The complete file path is "
+          + fullBlockFilePath);
+      throw new IOException();
+    }
+    return resultList;
+  }
+
+  /**
+   * this method compact the delete delta files.
+   * @param deleteDeltaFiles
+   * @param blockName
+   * @param fullBlockFilePath
+   * @return
+   */
+  public static Boolean startCompactionDeleteDeltaFiles(List<String> deleteDeltaFiles,
+      String blockName, String fullBlockFilePath) throws IOException {
+
+    DeleteDeltaBlockDetails deleteDeltaBlockDetails = null;
+    CarbonDeleteFilesDataReader dataReader = new CarbonDeleteFilesDataReader();
+    try {
+      deleteDeltaBlockDetails =
+              dataReader.getCompactedDeleteDeltaFileFromBlock(deleteDeltaFiles, blockName);
+    } catch (Exception e) {
+      String blockFilePath = fullBlockFilePath
+              .substring(0, fullBlockFilePath.lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR));
+      LOGGER.error("Error while getting the delete delta blocks in path " + blockFilePath);
+      throw new IOException();
+    }
+    CarbonDeleteDeltaWriterImpl carbonDeleteWriter =
+            new CarbonDeleteDeltaWriterImpl(fullBlockFilePath,
+                    FileFactory.getFileType(fullBlockFilePath));
+    try {
+      carbonDeleteWriter.write(deleteDeltaBlockDetails);
+    } catch (IOException e) {
+      LOGGER.error("Error while writing compacted delete delta file " + fullBlockFilePath);
+      throw new IOException();
+    }
+    return true;
+  }
+
+  public static Boolean updateStatusFile(
+          List<CarbonDataMergerUtilResult> updateDataMergerDetailsList, CarbonTable table,
+          String timestamp, SegmentUpdateStatusManager segmentUpdateStatusManager) {
+
+    List<SegmentUpdateDetails> segmentUpdateDetails =
+            new ArrayList<SegmentUpdateDetails>(updateDataMergerDetailsList.size());
+
+
+    // Check the list output.
+    for (CarbonDataMergerUtilResult carbonDataMergerUtilResult : updateDataMergerDetailsList) {
+      if (carbonDataMergerUtilResult.getCompactionStatus()) {
+        SegmentUpdateDetails tempSegmentUpdateDetails = new SegmentUpdateDetails();
+        tempSegmentUpdateDetails.setSegmentName(carbonDataMergerUtilResult.getSegmentName());
+        tempSegmentUpdateDetails.setBlockName(carbonDataMergerUtilResult.getBlockName());
+
+        for (SegmentUpdateDetails origDetails : segmentUpdateStatusManager
+                .getUpdateStatusDetails()) {
+          if (origDetails.getBlockName().equalsIgnoreCase(carbonDataMergerUtilResult.getBlockName())
+                  && origDetails.getSegmentName()
+                  .equalsIgnoreCase(carbonDataMergerUtilResult.getSegmentName())) {
+
+            tempSegmentUpdateDetails.setDeletedRowsInBlock(origDetails.getDeletedRowsInBlock());
+            tempSegmentUpdateDetails.setStatus(origDetails.getStatus());
+            break;
+          }
+        }
+
+        tempSegmentUpdateDetails.setDeleteDeltaStartTimestamp(
+                carbonDataMergerUtilResult.getDeleteDeltaStartTimestamp());
+        tempSegmentUpdateDetails
+              .setDeleteDeltaEndTimestamp(carbonDataMergerUtilResult.getDeleteDeltaEndTimestamp());
+
+        segmentUpdateDetails.add(tempSegmentUpdateDetails);
+      } else return false;
+    }
+
+    CarbonUpdateUtil.updateSegmentStatus(segmentUpdateDetails, table, timestamp, true);
+
+    // Update the Table Status.
+    String metaDataFilepath = table.getMetaDataFilepath();
+    AbsoluteTableIdentifier absoluteTableIdentifier = table.getAbsoluteTableIdentifier();
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+            .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+                    absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    String tableStatusPath = carbonTablePath.getTableStatusFilePath();
+
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+
+    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
+
+    boolean lockStatus = false;
+
+    try {
+      lockStatus = carbonLock.lockWithRetries();
+      if (lockStatus) {
+        LOGGER.info(
+                "Acquired lock for table" + table.getDatabaseName() + "." + table.getFactTableName()
+                        + " for table status updation");
+
+        LoadMetadataDetails[] listOfLoadFolderDetailsArray =
+                segmentStatusManager.readLoadMetadata(metaDataFilepath);
+
+        for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
+          if (loadMetadata.getLoadName().equalsIgnoreCase("0")) {
+            loadMetadata.setUpdateStatusFileName(
+                    CarbonUpdateUtil.getUpdateStatusFileName(timestamp));
+          }
+        }
+        try {
+          segmentStatusManager
+                  .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray);
+        } catch (IOException e) {
+          return false;
+        }
+      } else {
+        LOGGER.error("Not able to acquire the lock for Table status updation for table " + table
+                .getDatabaseName() + "." + table.getFactTableName());
+      }
+    } finally {
+      if (lockStatus) {
+        if (carbonLock.unlock()) {
+          LOGGER.info(
+                 "Table unlocked successfully after table status updation" + table.getDatabaseName()
+                          + "." + table.getFactTableName());
+        } else {
+          LOGGER.error(
+                  "Unable to unlock Table lock for table" + table.getDatabaseName() + "." + table
+                          .getFactTableName() + " during table status updation");
+        }
+      }
+    }
+    return true;
+  }
+
+  /**
+   * This will update the property of segments as major compacted.
+   * @param model
+   * @param changedSegDetails
+   */
+  public static void updateMajorCompactionPropertyInSegment(CarbonLoadModel model,
+      List<LoadMetadataDetails> changedSegDetails,
+      List<LoadMetadataDetails> preservedSegment) throws Exception {
+
+    String metadataPath = model.getCarbonDataLoadSchema().getCarbonTable().getMetaDataFilepath();
+    AbsoluteTableIdentifier absoluteTableIdentifier =
+            model.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+    LoadMetadataDetails[] details = segmentStatusManager.readLoadMetadata(metadataPath);
+    List<LoadMetadataDetails> originalList = Arrays.asList(details);
+    for (LoadMetadataDetails segment : changedSegDetails) {
+      if (preservedSegment.contains(segment)) {
+        continue;
+      }
+      originalList.get(originalList.indexOf(segment)).setMajorCompacted("true");
+
+    }
+
+
+    ICarbonLock carbonTableStatusLock = CarbonLockFactory.getCarbonLockObj(
+            model.getCarbonDataLoadSchema().getCarbonTable().getCarbonTableIdentifier(),
+            LockUsage.TABLE_STATUS_LOCK);
+
+    try {
+      if (carbonTableStatusLock.lockWithRetries()) {
+        LOGGER.info(
+            "Acquired lock for the table " + model.getDatabaseName() + "." + model.getTableName()
+                        + " for table status updation ");
+        CarbonTablePath carbonTablePath = CarbonStorePath
+                .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+                        absoluteTableIdentifier.getCarbonTableIdentifier());
+
+        segmentStatusManager.writeLoadDetailsIntoFile(carbonTablePath.getTableStatusFilePath(),
+                originalList.toArray(new LoadMetadataDetails[originalList.size()]));
+      } else {
+        LOGGER.error(
+                "Could not able to obtain lock for table" + model.getDatabaseName() + "." + model
+                        .getTableName() + "for table status updation");
+        throw new Exception("Failed to update the MajorCompactionStatus.");
+      }
+    } catch (IOException e) {
+      LOGGER.error("Error while writing metadata");
+      throw new Exception("Failed to update the MajorCompactionStatus." + e.getMessage());
+    } finally {
+      if (carbonTableStatusLock.unlock()) {
+        LOGGER.info(
+                "Table unlocked successfully after table status updation" + model.getDatabaseName()
+                        + "." + model.getTableName());
+      } else {
+        LOGGER.error("Unable to unlock Table lock for table" + model.getDatabaseName() + "." + model
+                .getTableName() + " during table status updation");
+      }
+    }
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtilResult.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtilResult.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtilResult.java
new file mode 100644
index 0000000..aa3d801
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtilResult.java
@@ -0,0 +1,33 @@
+/*
+ * 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.merger;
+
+import org.apache.carbondata.core.mutate.SegmentUpdateDetails;
+
+public final class CarbonDataMergerUtilResult extends SegmentUpdateDetails {
+  private boolean compactionStatus;
+
+  public boolean getCompactionStatus() {
+    return compactionStatus;
+  }
+
+  public void setCompactionStatus(Boolean status) {
+    compactionStatus = status;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
new file mode 100644
index 0000000..ebf3683
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
@@ -0,0 +1,401 @@
+/*
+ * 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.merger;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.SegmentProperties;
+import org.apache.carbondata.core.metadata.encoder.Encoding;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
+import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.processing.model.CarbonLoadModel;
+import org.apache.carbondata.processing.newflow.row.CarbonRow;
+import org.apache.carbondata.processing.sortandgroupby.exception.CarbonSortKeyAndGroupByException;
+import org.apache.carbondata.processing.sortandgroupby.sortdata.SortDataRows;
+import org.apache.carbondata.processing.sortandgroupby.sortdata.SortIntermediateFileMerger;
+import org.apache.carbondata.processing.sortandgroupby.sortdata.SortParameters;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
+import org.apache.carbondata.processing.store.CarbonFactHandler;
+import org.apache.carbondata.processing.store.CarbonFactHandlerFactory;
+import org.apache.carbondata.processing.store.SingleThreadFinalSortFilesMerger;
+import org.apache.carbondata.processing.store.writer.exception.CarbonDataWriterException;
+import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
+
+/**
+ * This class will process the query result and convert the data
+ * into a format compatible for data load
+ */
+public class CompactionResultSortProcessor extends AbstractResultProcessor {
+
+  /**
+   * LOGGER
+   */
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(CompactionResultSortProcessor.class.getName());
+  /**
+   * carbon load model that contains all the required information for load
+   */
+  private CarbonLoadModel carbonLoadModel;
+  /**
+   * carbon table
+   */
+  private CarbonTable carbonTable;
+  /**
+   * sortDataRows instance for sorting each row read ad writing to sort temp file
+   */
+  private SortDataRows sortDataRows;
+  /**
+   * final merger for merge sort
+   */
+  private SingleThreadFinalSortFilesMerger finalMerger;
+  /**
+   * data handler VO object
+   */
+  private CarbonFactHandler dataHandler;
+  /**
+   * segment properties for getting dimension cardinality and other required information of a block
+   */
+  private SegmentProperties segmentProperties;
+  /**
+   * compaction type to decide whether taskID need to be extracted from carbondata files
+   */
+  private CompactionType compactionType;
+  /**
+   * boolean mapping for no dictionary columns in schema
+   */
+  private boolean[] noDictionaryColMapping;
+  /**
+   * agg type defined for measures
+   */
+  private char[] aggType;
+  /**
+   * segment id
+   */
+  private String segmentId;
+  /**
+   * temp store location to be sued during data load
+   */
+  private String tempStoreLocation;
+  /**
+   * table name
+   */
+  private String tableName;
+  /**
+   * no dictionary column count in schema
+   */
+  private int noDictionaryCount;
+  /**
+   * total count of measures in schema
+   */
+  private int measureCount;
+  /**
+   * dimension count excluding complex dimension and no dictionary column count
+   */
+  private int dimensionColumnCount;
+  /**
+   * whether the allocated tasks has any record
+   */
+  private boolean isRecordFound;
+
+  /**
+   * @param carbonLoadModel
+   * @param carbonTable
+   * @param segmentProperties
+   * @param compactionType
+   * @param tableName
+   */
+  public CompactionResultSortProcessor(CarbonLoadModel carbonLoadModel, CarbonTable carbonTable,
+      SegmentProperties segmentProperties, CompactionType compactionType, String tableName) {
+    this.carbonLoadModel = carbonLoadModel;
+    this.carbonTable = carbonTable;
+    this.segmentProperties = segmentProperties;
+    this.segmentId = carbonLoadModel.getSegmentId();
+    this.compactionType = compactionType;
+    this.tableName = tableName;
+  }
+
+  /**
+   * This method will iterate over the query result and convert it into a format compatible
+   * for data loading
+   *
+   * @param resultIteratorList
+   */
+  public boolean execute(List<RawResultIterator> resultIteratorList) {
+    boolean isCompactionSuccess = false;
+    try {
+      initTempStoreLocation();
+      initSortDataRows();
+      initAggType();
+      processResult(resultIteratorList);
+      // After delete command, if no records are fetched from one split,
+      // below steps are not required to be initialized.
+      if (isRecordFound) {
+        initializeFinalThreadMergerForMergeSort();
+        initDataHandler();
+        readAndLoadDataFromSortTempFiles();
+      }
+      isCompactionSuccess = true;
+    } catch (Exception e) {
+      LOGGER.error(e, "Compaction failed: " + e.getMessage());
+    } finally {
+      // clear temp files and folders created during compaction
+      deleteTempStoreLocation();
+    }
+    return isCompactionSuccess;
+  }
+
+  /**
+   * This method will clean up the local folders and files created during compaction process
+   */
+  private void deleteTempStoreLocation() {
+    if (null != tempStoreLocation) {
+      try {
+        CarbonUtil.deleteFoldersAndFiles(new File[] { new File(tempStoreLocation) });
+      } catch (IOException | InterruptedException e) {
+        LOGGER.error("Problem deleting local folders during compaction: " + e.getMessage());
+      }
+    }
+  }
+
+  /**
+   * This method will iterate over the query result and perform row sorting operation
+   *
+   * @param resultIteratorList
+   */
+  private void processResult(List<RawResultIterator> resultIteratorList)
+      throws Exception {
+    for (RawResultIterator resultIterator : resultIteratorList) {
+      while (resultIterator.hasNext()) {
+        addRowForSorting(prepareRowObjectForSorting(resultIterator.next()));
+        isRecordFound = true;
+      }
+    }
+    try {
+      sortDataRows.startSorting();
+    } catch (CarbonSortKeyAndGroupByException e) {
+      LOGGER.error(e);
+      throw new Exception("Problem loading data during compaction: " + e.getMessage());
+    }
+  }
+
+  /**
+   * This method will prepare the data from raw object that will take part in sorting
+   *
+   * @param row
+   * @return
+   */
+  private Object[] prepareRowObjectForSorting(Object[] row) {
+    ByteArrayWrapper wrapper = (ByteArrayWrapper) row[0];
+    // ByteBuffer[] noDictionaryBuffer = new ByteBuffer[noDictionaryCount];
+    List<CarbonDimension> dimensions = segmentProperties.getDimensions();
+    Object[] preparedRow = new Object[dimensions.size() + measureCount];
+    // convert the dictionary from MDKey to surrogate key
+    byte[] dictionaryKey = wrapper.getDictionaryKey();
+    long[] keyArray = segmentProperties.getDimensionKeyGenerator().getKeyArray(dictionaryKey);
+    Object[] dictionaryValues = new Object[dimensionColumnCount + measureCount];
+    for (int i = 0; i < keyArray.length; i++) {
+      dictionaryValues[i] = Long.valueOf(keyArray[i]).intValue();
+    }
+    int noDictionaryIndex = 0;
+    int dictionaryIndex = 0;
+    for (int i = 0; i < dimensions.size(); i++) {
+      CarbonDimension dims = dimensions.get(i);
+      if (dims.hasEncoding(Encoding.DICTIONARY)) {
+        // dictionary
+        preparedRow[i] = dictionaryValues[dictionaryIndex++];
+      } else {
+        // no dictionary dims
+        preparedRow[i] = wrapper.getNoDictionaryKeyByIndex(noDictionaryIndex++);
+      }
+    }
+    // fill all the measures
+    // measures will always start from 1st index in the row object array
+    int measureIndexInRow = 1;
+    for (int i = 0; i < measureCount; i++) {
+      preparedRow[dimensionColumnCount + i] =
+          getConvertedMeasureValue(row[measureIndexInRow++], aggType[i]);
+    }
+    return preparedRow;
+  }
+
+  /**
+   * This method will convert the spark decimal to java big decimal type
+   *
+   * @param value
+   * @param aggType
+   * @return
+   */
+  private Object getConvertedMeasureValue(Object value, char aggType) {
+    switch (aggType) {
+      case CarbonCommonConstants.BIG_DECIMAL_MEASURE:
+        value = ((org.apache.spark.sql.types.Decimal) value).toJavaBigDecimal();
+        return value;
+      default:
+        return value;
+    }
+  }
+
+  /**
+   * This method will read sort temp files, perform merge sort and add it to store for data loading
+   */
+  private void readAndLoadDataFromSortTempFiles() throws Exception {
+    try {
+      finalMerger.startFinalMerge();
+      while (finalMerger.hasNext()) {
+        Object[] rowRead = finalMerger.next();
+        CarbonRow row = new CarbonRow(rowRead);
+        // convert the row from surrogate key to MDKey
+        Object[] outputRow = CarbonDataProcessorUtil
+            .convertToMDKeyAndFillRow(row, segmentProperties, measureCount, noDictionaryCount,
+                segmentProperties.getComplexDimensions().size());
+        dataHandler.addDataToStore(outputRow);
+      }
+      dataHandler.finish();
+    } catch (CarbonDataWriterException e) {
+      LOGGER.error(e);
+      throw new Exception("Problem loading data during compaction: " + e.getMessage());
+    } catch (Exception e) {
+      LOGGER.error(e);
+      throw new Exception("Problem loading data during compaction: " + e.getMessage());
+    } finally {
+      if (null != dataHandler) {
+        try {
+          dataHandler.closeHandler();
+        } catch (CarbonDataWriterException e) {
+          LOGGER.error(e);
+          throw new Exception("Problem loading data during compaction: " + e.getMessage());
+        }
+      }
+    }
+  }
+
+  /**
+   * add row to a temp array which will we written to a sort temp file after sorting
+   *
+   * @param row
+   */
+  private void addRowForSorting(Object[] row) throws Exception {
+    try {
+      // prepare row array using RemoveDictionaryUtil class
+      sortDataRows.addRow(row);
+    } catch (CarbonSortKeyAndGroupByException e) {
+      LOGGER.error(e);
+      throw new Exception("Row addition for sorting failed during compaction: " + e.getMessage());
+    }
+  }
+
+  /**
+   * create an instance of sort data rows
+   */
+  private void initSortDataRows() throws Exception {
+    measureCount = carbonTable.getMeasureByTableName(tableName).size();
+    List<CarbonDimension> dimensions = carbonTable.getDimensionByTableName(tableName);
+    noDictionaryColMapping = new boolean[dimensions.size()];
+    int i = 0;
+    for (CarbonDimension dimension : dimensions) {
+      if (CarbonUtil.hasEncoding(dimension.getEncoder(), Encoding.DICTIONARY)) {
+        i++;
+        continue;
+      }
+      noDictionaryColMapping[i++] = true;
+      noDictionaryCount++;
+    }
+    dimensionColumnCount = dimensions.size();
+    SortParameters parameters = createSortParameters();
+    SortIntermediateFileMerger intermediateFileMerger = new SortIntermediateFileMerger(parameters);
+    // TODO: Now it is only supported onheap merge, but we can have unsafe merge
+    // as well by using UnsafeSortDataRows.
+    this.sortDataRows = new SortDataRows(parameters, intermediateFileMerger);
+    try {
+      this.sortDataRows.initialize();
+    } catch (CarbonSortKeyAndGroupByException e) {
+      LOGGER.error(e);
+      throw new Exception(
+          "Error initializing sort data rows object during compaction: " + e.getMessage());
+    }
+  }
+
+  /**
+   * This method will create the sort parameters VO object
+   *
+   * @return
+   */
+  private SortParameters createSortParameters() {
+    SortParameters parameters = SortParameters
+        .createSortParameters(carbonLoadModel.getDatabaseName(), tableName, dimensionColumnCount,
+            segmentProperties.getComplexDimensions().size(), measureCount, noDictionaryCount,
+            carbonLoadModel.getPartitionId(), segmentId, carbonLoadModel.getTaskNo(),
+            noDictionaryColMapping);
+    return parameters;
+  }
+
+  /**
+   * create an instance of finalThread merger which will perform merge sort on all the
+   * sort temp files
+   */
+  private void initializeFinalThreadMergerForMergeSort() {
+    String sortTempFileLocation = tempStoreLocation + CarbonCommonConstants.FILE_SEPARATOR
+        + CarbonCommonConstants.SORT_TEMP_FILE_LOCATION;
+    finalMerger =
+        new SingleThreadFinalSortFilesMerger(sortTempFileLocation, tableName, dimensionColumnCount,
+            segmentProperties.getComplexDimensions().size(), measureCount, noDictionaryCount,
+            aggType, noDictionaryColMapping);
+  }
+
+  /**
+   * initialise carbon data writer instance
+   */
+  private void initDataHandler() throws Exception {
+    CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel
+        .getCarbonFactDataHandlerModel(carbonLoadModel, carbonTable, segmentProperties, tableName,
+            tempStoreLocation);
+    setDataFileAttributesInModel(carbonLoadModel, compactionType, carbonTable,
+        carbonFactDataHandlerModel);
+    dataHandler = CarbonFactHandlerFactory.createCarbonFactHandler(carbonFactDataHandlerModel,
+        CarbonFactHandlerFactory.FactHandlerType.COLUMNAR);
+    try {
+      dataHandler.initialise();
+    } catch (CarbonDataWriterException e) {
+      LOGGER.error(e);
+      throw new Exception("Problem initialising data handler during compaction: " + e.getMessage());
+    }
+  }
+
+  /**
+   * initialise temporary store location
+   */
+  private void initTempStoreLocation() {
+    tempStoreLocation = CarbonDataProcessorUtil
+        .getLocalDataFolderLocation(carbonLoadModel.getDatabaseName(), tableName,
+            carbonLoadModel.getTaskNo(), carbonLoadModel.getPartitionId(), segmentId, false);
+  }
+
+  /**
+   * initialise aggregation type for measures for their storage format
+   */
+  private void initAggType() {
+    aggType = CarbonDataProcessorUtil.initAggType(carbonTable, tableName, measureCount);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionType.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionType.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionType.java
new file mode 100644
index 0000000..6b9c80a
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionType.java
@@ -0,0 +1,31 @@
+/*
+ * 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.merger;
+
+/**
+ * This enum is used to define the types of Compaction.
+ * We have 3 types. one is Minor another is Major and
+ * finally a compaction done after UPDATE-DELETE operation
+ * called IUD compaction.
+ */
+public enum CompactionType {
+    MINOR_COMPACTION,
+    MAJOR_COMPACTION,
+    IUD_UPDDEL_DELTA_COMPACTION,
+    IUD_DELETE_DELTA_COMPACTION,
+    NONE
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/NodeBlockRelation.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/NodeBlockRelation.java b/processing/src/main/java/org/apache/carbondata/processing/merger/NodeBlockRelation.java
new file mode 100644
index 0000000..ecf4408
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/NodeBlockRelation.java
@@ -0,0 +1,58 @@
+/*
+ * 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.merger;
+
+import org.apache.carbondata.core.datastore.block.Distributable;
+
+/**
+ * Block to Node mapping
+ */
+public class NodeBlockRelation implements Comparable<NodeBlockRelation> {
+
+  private final Distributable block;
+  private final String node;
+
+  public NodeBlockRelation(Distributable block, String node) {
+    this.block = block;
+    this.node = node;
+
+  }
+
+  public Distributable getBlock() {
+    return block;
+  }
+
+  public String getNode() {
+    return node;
+  }
+
+  @Override public int compareTo(NodeBlockRelation obj) {
+    return this.getNode().compareTo(obj.getNode());
+  }
+
+  @Override public boolean equals(Object obj) {
+    if (!(obj instanceof NodeBlockRelation)) {
+      return false;
+    }
+    NodeBlockRelation o = (NodeBlockRelation) obj;
+    return node.equals(o.node);
+  }
+
+  @Override public int hashCode() {
+    return node.hashCode();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/NodeMultiBlockRelation.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/NodeMultiBlockRelation.java b/processing/src/main/java/org/apache/carbondata/processing/merger/NodeMultiBlockRelation.java
new file mode 100644
index 0000000..ec2ddaf
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/NodeMultiBlockRelation.java
@@ -0,0 +1,57 @@
+/*
+ * 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.merger;
+
+import java.util.List;
+
+import org.apache.carbondata.core.datastore.block.Distributable;
+
+public class NodeMultiBlockRelation implements Comparable<NodeMultiBlockRelation> {
+
+  private final List<Distributable> blocks;
+  private final String node;
+
+  public NodeMultiBlockRelation(String node, List<Distributable> blocks) {
+    this.node = node;
+    this.blocks = blocks;
+
+  }
+
+  public List<Distributable> getBlocks() {
+    return blocks;
+  }
+
+  public String getNode() {
+    return node;
+  }
+
+  @Override public int compareTo(NodeMultiBlockRelation obj) {
+    return this.blocks.size() - obj.getBlocks().size();
+  }
+
+  @Override public boolean equals(Object obj) {
+    if (!(obj instanceof NodeMultiBlockRelation)) {
+      return false;
+    }
+    NodeMultiBlockRelation o = (NodeMultiBlockRelation) obj;
+    return blocks.equals(o.blocks) && node.equals(o.node);
+  }
+
+  @Override public int hashCode() {
+    return blocks.hashCode() + node.hashCode();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/cc59b247/processing/src/main/java/org/apache/carbondata/processing/merger/RowResultMergerProcessor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/RowResultMergerProcessor.java b/processing/src/main/java/org/apache/carbondata/processing/merger/RowResultMergerProcessor.java
new file mode 100644
index 0000000..b7aa32c
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/RowResultMergerProcessor.java
@@ -0,0 +1,233 @@
+/*
+ * 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.merger;
+
+import java.io.File;
+import java.util.AbstractQueue;
+import java.util.Comparator;
+import java.util.List;
+import java.util.PriorityQueue;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.SegmentProperties;
+import org.apache.carbondata.core.keygenerator.KeyGenException;
+import org.apache.carbondata.core.metadata.CarbonMetadata;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
+import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
+import org.apache.carbondata.core.util.ByteUtil;
+import org.apache.carbondata.processing.merger.exeception.SliceMergerException;
+import org.apache.carbondata.processing.model.CarbonLoadModel;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerColumnar;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
+import org.apache.carbondata.processing.store.CarbonFactHandler;
+import org.apache.carbondata.processing.store.writer.exception.CarbonDataWriterException;
+
+/**
+ * This is the Merger class responsible for the merging of the segments.
+ */
+public class RowResultMergerProcessor extends AbstractResultProcessor {
+
+  private CarbonFactHandler dataHandler;
+  private SegmentProperties segprop;
+  /**
+   * record holder heap
+   */
+  private AbstractQueue<RawResultIterator> recordHolderHeap;
+
+  private TupleConversionAdapter tupleConvertor;
+
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(RowResultMergerProcessor.class.getName());
+
+  public RowResultMergerProcessor(String databaseName,
+      String tableName, SegmentProperties segProp, String tempStoreLocation,
+      CarbonLoadModel loadModel, CompactionType compactionType) {
+    this.segprop = segProp;
+    if (!new File(tempStoreLocation).mkdirs()) {
+      LOGGER.error("Error while new File(tempStoreLocation).mkdirs() ");
+    }
+    CarbonTable carbonTable = CarbonMetadata.getInstance()
+            .getCarbonTable(databaseName + CarbonCommonConstants.UNDERSCORE + tableName);
+    CarbonFactDataHandlerModel carbonFactDataHandlerModel = CarbonFactDataHandlerModel
+        .getCarbonFactDataHandlerModel(loadModel, carbonTable, segProp, tableName,
+            tempStoreLocation);
+    setDataFileAttributesInModel(loadModel, compactionType, carbonTable,
+        carbonFactDataHandlerModel);
+    carbonFactDataHandlerModel.setCompactionFlow(true);
+    dataHandler = new CarbonFactDataHandlerColumnar(carbonFactDataHandlerModel);
+    tupleConvertor = new TupleConversionAdapter(segProp);
+  }
+
+  private void initRecordHolderHeap(List<RawResultIterator> rawResultIteratorList) {
+    // create the List of RawResultIterator.
+    recordHolderHeap = new PriorityQueue<RawResultIterator>(rawResultIteratorList.size(),
+        new RowResultMergerProcessor.CarbonMdkeyComparator());
+  }
+
+  /**
+   * Merge function
+   *
+   */
+  public boolean execute(List<RawResultIterator> resultIteratorList) {
+    initRecordHolderHeap(resultIteratorList);
+    boolean mergeStatus = false;
+    int index = 0;
+    boolean isDataPresent = false;
+    try {
+
+      // add all iterators to the queue
+      for (RawResultIterator leaftTupleIterator : resultIteratorList) {
+        this.recordHolderHeap.add(leaftTupleIterator);
+        index++;
+      }
+      RawResultIterator iterator = null;
+      while (index > 1) {
+        // iterator the top record
+        iterator = this.recordHolderHeap.poll();
+        Object[] convertedRow = iterator.next();
+        if (null == convertedRow) {
+          index--;
+          continue;
+        }
+        if (!isDataPresent) {
+          dataHandler.initialise();
+          isDataPresent = true;
+        }
+        // get the mdkey
+        addRow(convertedRow);
+        // if there is no record in the leaf and all then decrement the
+        // index
+        if (!iterator.hasNext()) {
+          index--;
+          continue;
+        }
+        // add record to heap
+        this.recordHolderHeap.add(iterator);
+      }
+      // if record holder is not empty then iterator the slice holder from
+      // heap
+      iterator = this.recordHolderHeap.poll();
+      while (true) {
+        Object[] convertedRow = iterator.next();
+        if (null == convertedRow) {
+          break;
+        }
+        // do it only once
+        if (!isDataPresent) {
+          dataHandler.initialise();
+          isDataPresent = true;
+        }
+        addRow(convertedRow);
+        // check if leaf contains no record
+        if (!iterator.hasNext()) {
+          break;
+        }
+      }
+      if (isDataPresent)
+      {
+        this.dataHandler.finish();
+      }
+      mergeStatus = true;
+    } catch (Exception e) {
+      LOGGER.error(e, e.getMessage());
+      LOGGER.error("Exception in compaction merger " + e.getMessage());
+      mergeStatus = false;
+    } finally {
+      try {
+        if (isDataPresent) {
+          this.dataHandler.closeHandler();
+        }
+      } catch (CarbonDataWriterException e) {
+        LOGGER.error("Exception while closing the handler in compaction merger " + e.getMessage());
+        mergeStatus = false;
+      }
+    }
+
+    return mergeStatus;
+  }
+
+  /**
+   * Below method will be used to add sorted row
+   *
+   * @throws SliceMergerException
+   */
+  private void addRow(Object[] carbonTuple) throws SliceMergerException {
+    Object[] rowInWritableFormat;
+
+    rowInWritableFormat = tupleConvertor.getObjectArray(carbonTuple);
+    try {
+      this.dataHandler.addDataToStore(rowInWritableFormat);
+    } catch (CarbonDataWriterException e) {
+      throw new SliceMergerException("Problem in merging the slice", e);
+    }
+  }
+
+  /**
+   * Comparator class for comparing 2 raw row result.
+   */
+  private class CarbonMdkeyComparator implements Comparator<RawResultIterator> {
+
+    @Override public int compare(RawResultIterator o1, RawResultIterator o2) {
+
+      Object[] row1 = new Object[0];
+      Object[] row2 = new Object[0];
+      try {
+        row1 = o1.fetchConverted();
+        row2 = o2.fetchConverted();
+      } catch (KeyGenException e) {
+        LOGGER.error(e.getMessage());
+      }
+      if (null == row1 || null == row2) {
+        return 0;
+      }
+      ByteArrayWrapper key1 = (ByteArrayWrapper) row1[0];
+      ByteArrayWrapper key2 = (ByteArrayWrapper) row2[0];
+      int compareResult = 0;
+      int[] columnValueSizes = segprop.getEachDimColumnValueSize();
+      int dictionaryKeyOffset = 0;
+      byte[] dimCols1 = key1.getDictionaryKey();
+      byte[] dimCols2 = key2.getDictionaryKey();
+      int noDicIndex = 0;
+      for (int eachColumnValueSize : columnValueSizes) {
+        // case of dictionary cols
+        if (eachColumnValueSize > 0) {
+
+          compareResult = ByteUtil.UnsafeComparer.INSTANCE
+              .compareTo(dimCols1, dictionaryKeyOffset, eachColumnValueSize, dimCols2,
+                  dictionaryKeyOffset, eachColumnValueSize);
+          dictionaryKeyOffset += eachColumnValueSize;
+        } else { // case of no dictionary
+
+          byte[] noDictionaryDim1 = key1.getNoDictionaryKeyByIndex(noDicIndex);
+          byte[] noDictionaryDim2 = key2.getNoDictionaryKeyByIndex(noDicIndex);
+          compareResult =
+              ByteUtil.UnsafeComparer.INSTANCE.compareTo(noDictionaryDim1, noDictionaryDim2);
+          noDicIndex++;
+
+        }
+        if (0 != compareResult) {
+          return compareResult;
+        }
+      }
+      return 0;
+    }
+  }
+
+}



Mime
View raw message