carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chenliang...@apache.org
Subject [21/52] [partial] incubator-carbondata git commit: move core package
Date Mon, 16 Jan 2017 14:52:58 GMT
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorder.java b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorder.java
new file mode 100644
index 0000000..fd9c0ef
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorder.java
@@ -0,0 +1,36 @@
+/*
+ * 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.stats;
+
+/**
+ * interface will be used to record and log the query statistics
+ */
+public interface QueryStatisticsRecorder {
+
+  void recordStatistics(QueryStatistic statistic);
+
+  void logStatistics();
+
+  void logStatisticsAsTableExecutor();
+
+  void recordStatisticsForDriver(QueryStatistic statistic, String queryId);
+
+  void logStatisticsAsTableDriver();
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderDummy.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderDummy.java b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderDummy.java
new file mode 100644
index 0000000..a751936
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderDummy.java
@@ -0,0 +1,68 @@
+/*
+ * 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.stats;
+
+import java.io.Serializable;
+
+/**
+ * Class will be used to record and log the query statistics
+ */
+public class QueryStatisticsRecorderDummy implements QueryStatisticsRecorder,Serializable {
+
+  /**
+   * serialization version
+   */
+  private static final long serialVersionUID = -5719752001674467864L;
+
+  public QueryStatisticsRecorderDummy() {
+
+  }
+
+  /**
+   * Below method will be used to add the statistics
+   *
+   * @param statistic
+   */
+  public synchronized void recordStatistics(QueryStatistic statistic) {
+
+  }
+
+  /**
+   * Below method will be used to log the statistic
+   */
+  public void logStatistics() {
+
+  }
+
+  /**
+   * Below method will be used to show statistic log as table
+   */
+  public void logStatisticsAsTableExecutor() {
+
+  }
+
+  public void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
+
+  }
+
+  public void logStatisticsAsTableDriver() {
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderImpl.java b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderImpl.java
new file mode 100644
index 0000000..9cfeab4
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/stats/QueryStatisticsRecorderImpl.java
@@ -0,0 +1,176 @@
+/*
+ * 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.stats;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+
+import static org.apache.carbondata.core.util.CarbonUtil.printLine;
+
+/**
+ * Class will be used to record and log the query statistics
+ */
+public class QueryStatisticsRecorderImpl implements QueryStatisticsRecorder,Serializable {
+
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(QueryStatisticsRecorderImpl.class.getName());
+
+  /**
+   * serialization version
+   */
+  private static final long serialVersionUID = -5719752001674467864L;
+
+  /**
+   * list for statistics to record time taken
+   * by each phase of the query for example aggregation
+   * scanning,block loading time etc.
+   */
+  private List<QueryStatistic> queryStatistics;
+
+  /**
+   * query with taskd
+   */
+  private String queryIWthTask;
+
+  public QueryStatisticsRecorderImpl(String queryId) {
+    queryStatistics = new ArrayList<QueryStatistic>();
+    this.queryIWthTask = queryId;
+  }
+
+  /**
+   * Below method will be used to add the statistics
+   *
+   * @param statistic
+   */
+  public synchronized void recordStatistics(QueryStatistic statistic) {
+    queryStatistics.add(statistic);
+  }
+
+  /**
+   * Below method will be used to log the statistic
+   */
+  public void logStatistics() {
+    for (QueryStatistic statistic : queryStatistics) {
+      LOGGER.statistic(statistic.getStatistics(queryIWthTask));
+    }
+  }
+
+  /**
+   * Below method will be used to show statistic log as table
+   */
+  public void logStatisticsAsTableExecutor() {
+    String tableInfo = collectExecutorStatistics();
+    if (null != tableInfo) {
+      LOGGER.statistic(tableInfo);
+    }
+  }
+
+  /**
+   * Below method will parse queryStatisticsMap and put time into table
+   */
+  public String collectExecutorStatistics() {
+    String load_blocks_time = "";
+    String scan_blocks_time = "";
+    String scan_blocks_num = "";
+    String load_dictionary_time = "";
+    String result_size = "";
+    String total_executor_time = "";
+    String splitChar = " ";
+    String total_blocklet = "";
+    String valid_scan_blocklet = "";
+    try {
+      for (QueryStatistic statistic : queryStatistics) {
+        switch (statistic.getMessage()) {
+          case QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR:
+            load_blocks_time += statistic.getTimeTaken() + splitChar;
+            break;
+          case QueryStatisticsConstants.SCAN_BLOCKS_TIME:
+            scan_blocks_time += statistic.getTimeTaken() + splitChar;
+            break;
+          case QueryStatisticsConstants.SCAN_BLOCKS_NUM:
+            scan_blocks_num += statistic.getCount() + splitChar;
+            break;
+          case QueryStatisticsConstants.LOAD_DICTIONARY:
+            load_dictionary_time += statistic.getTimeTaken() + splitChar;
+            break;
+          case QueryStatisticsConstants.RESULT_SIZE:
+            result_size += statistic.getCount() + splitChar;
+            break;
+          case QueryStatisticsConstants.EXECUTOR_PART:
+            total_executor_time += statistic.getTimeTaken() + splitChar;
+            break;
+          case QueryStatisticsConstants.TOTAL_BLOCKLET_NUM:
+            total_blocklet = statistic.getCount() + splitChar;
+            break;
+          case QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM:
+            valid_scan_blocklet = statistic.getCount() + splitChar;
+            break;
+          default:
+            break;
+        }
+      }
+      String headers = "task_id,load_blocks_time,load_dictionary_time,scan_blocks_time," +
+          "total_executor_time,scan_blocks_num,total_blocklet," +
+          "valid_scan_blocklet,result_size";
+      List<String> values = new ArrayList<String>();
+      values.add(queryIWthTask);
+      values.add(load_blocks_time);
+      values.add(load_dictionary_time);
+      values.add(scan_blocks_time);
+      values.add(total_executor_time);
+      values.add(scan_blocks_num);
+      values.add(total_blocklet);
+      values.add(valid_scan_blocklet);
+      values.add(result_size);
+      StringBuilder tableInfo = new StringBuilder();
+      String[] columns = headers.split(",");
+      String line = "";
+      String hearLine = "";
+      String valueLine = "";
+      for (int i = 0; i < columns.length; i++) {
+        int len = Math.max(columns[i].length(), values.get(i).length());
+        line += "+" + printLine("-", len);
+        hearLine += "|" + printLine(" ", len - columns[i].length()) + columns[i];
+        valueLine += "|" + printLine(" ", len - values.get(i).length()) + values.get(i);
+      }
+      // struct table info
+      tableInfo.append(line + "+").append("\n");
+      tableInfo.append(hearLine + "|").append("\n");
+      tableInfo.append(line + "+").append("\n");
+      tableInfo.append(valueLine + "|").append("\n");
+      tableInfo.append(line + "+").append("\n");
+      return "Print query statistic for each task id:" + "\n" + tableInfo.toString();
+    } catch (Exception ex) {
+      return "Put statistics into table failed, catch exception: " + ex.getMessage();
+    }
+  }
+
+  public void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
+
+  }
+
+  public void logStatisticsAsTableDriver() {
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/statusmanager/LoadMetadataDetails.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/LoadMetadataDetails.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/LoadMetadataDetails.java
new file mode 100644
index 0000000..70043d4
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/LoadMetadataDetails.java
@@ -0,0 +1,344 @@
+/*
+ * 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.statusmanager;
+
+import java.io.Serializable;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+
+public class LoadMetadataDetails implements Serializable {
+
+  private static final long serialVersionUID = 1106104914918491724L;
+  private String timestamp;
+  private String loadStatus;
+  private String loadName;
+  private String partitionCount;
+  private String isDeleted = CarbonCommonConstants.KEYWORD_FALSE;
+
+  // update delta end timestamp
+  private String updateDeltaEndTimestamp = "";
+
+  // update delta start timestamp
+  private String updateDeltaStartTimestamp = "";
+
+  // this will represent the update status file name at that point of time.
+  private String updateStatusFileName = "";
+
+  /**
+   * LOGGER
+   */
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(LoadMetadataDetails.class.getName());
+
+  // dont remove static as the write will fail.
+  private static final SimpleDateFormat parser =
+      new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP);
+  /**
+   * Segment modification or deletion time stamp
+   */
+  private String modificationOrdeletionTimesStamp;
+  private String loadStartTime;
+
+  private String mergedLoadName;
+  /**
+   * visibility is used to determine whether to the load is visible or not.
+   */
+  private String visibility = "true";
+
+  /**
+   * To know if the segment is a major compacted segment or not.
+   */
+  private String majorCompacted;
+
+  public String getPartitionCount() {
+    return partitionCount;
+  }
+
+  public void setPartitionCount(String partitionCount) {
+    this.partitionCount = partitionCount;
+  }
+
+  public long getLoadEndTime() {
+    return convertTimeStampToLong(timestamp);
+  }
+
+  public void setLoadEndTime(long timestamp) {
+    this.timestamp = getTimeStampConvertion(timestamp);;
+  }
+
+  public String getLoadStatus() {
+    return loadStatus;
+  }
+
+  public void setLoadStatus(String loadStatus) {
+    this.loadStatus = loadStatus;
+  }
+
+  public String getLoadName() {
+    return loadName;
+  }
+
+  public void setLoadName(String loadName) {
+    this.loadName = loadName;
+  }
+
+  /**
+   * @return the modificationOrdeletionTimesStamp
+   */
+  public long getModificationOrdeletionTimesStamp() {
+    if(null == modificationOrdeletionTimesStamp) {
+      return 0;
+    }
+    return convertTimeStampToLong(modificationOrdeletionTimesStamp);
+  }
+
+  /**
+   * @param modificationOrdeletionTimesStamp the modificationOrdeletionTimesStamp to set
+   */
+  public void setModificationOrdeletionTimesStamp(long modificationOrdeletionTimesStamp) {
+    this.modificationOrdeletionTimesStamp =
+        getTimeStampConvertion(modificationOrdeletionTimesStamp);
+  }
+
+  /* (non-Javadoc)
+   * @see java.lang.Object#hashCode()
+   */
+  @Override public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + ((loadName == null) ? 0 : loadName.hashCode());
+    return result;
+  }
+
+  /* (non-Javadoc)
+   * @see java.lang.Object#equals(java.lang.Object)
+   */
+  @Override public boolean equals(Object obj) {
+    if (obj == null) {
+      return false;
+
+    }
+    if (!(obj instanceof LoadMetadataDetails)) {
+      return false;
+    }
+    LoadMetadataDetails other = (LoadMetadataDetails) obj;
+    if (loadName == null) {
+      if (other.loadName != null) {
+        return false;
+      }
+    } else if (!loadName.equals(other.loadName)) {
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * @return the startLoadTime
+   */
+  public long getLoadStartTime() {
+    return convertTimeStampToLong(loadStartTime);
+  }
+
+  /**
+   * return loadStartTime
+   *
+   * @return
+   */
+  public long getLoadStartTimeAsLong() {
+    return (!loadStartTime.isEmpty()) ? getTimeStamp(loadStartTime) : 0;
+  }
+
+  /**
+   * This method will convert a given timestamp to long value and then to string back
+   *
+   * @param factTimeStamp
+   * @return
+   */
+  private long convertTimeStampToLong(String factTimeStamp) {
+    SimpleDateFormat parser = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_MILLIS);
+    Date dateToStr = null;
+    try {
+      dateToStr = parser.parse(factTimeStamp);
+      return dateToStr.getTime();
+    } catch (ParseException e) {
+      LOGGER.error("Cannot convert" + factTimeStamp + " to Time/Long type value" + e.getMessage());
+      parser = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP);
+      try {
+        dateToStr = parser.parse(factTimeStamp);
+        return dateToStr.getTime();
+      } catch (ParseException e1) {
+        LOGGER
+            .error("Cannot convert" + factTimeStamp + " to Time/Long type value" + e1.getMessage());
+        return 0;
+      }
+    }
+  }
+
+  /**
+   * returns load start time as long value
+   *
+   * @param loadStartTime
+   * @return
+   */
+  public Long getTimeStamp(String loadStartTime) {
+    Date dateToStr = null;
+    try {
+      dateToStr = parser.parse(loadStartTime);
+      return dateToStr.getTime() * 1000;
+    } catch (ParseException e) {
+      LOGGER.error("Cannot convert" + loadStartTime + " to Time/Long type value" + e.getMessage());
+      return null;
+    }
+  }
+
+  private String getTimeStampConvertion(long time) {
+    SimpleDateFormat sdf = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP_MILLIS);
+    return sdf.format(time);
+  }
+
+  /**
+   * @param loadStartTime
+   */
+  public void setLoadStartTime(long loadStartTime) {
+    this.loadStartTime = getTimeStampConvertion(loadStartTime);
+  }
+
+  /**
+   * @return the mergedLoadName
+   */
+  public String getMergedLoadName() {
+    return mergedLoadName;
+  }
+
+  /**
+   * @param mergedLoadName the mergedLoadName to set
+   */
+  public void setMergedLoadName(String mergedLoadName) {
+    this.mergedLoadName = mergedLoadName;
+  }
+
+  /**
+   * @return the visibility
+   */
+  public String getVisibility() {
+    return visibility;
+  }
+
+  /**
+   * @param visibility the visibility to set
+   */
+  public void setVisibility(String visibility) {
+    this.visibility = visibility;
+  }
+
+  /**
+   * Return true if it is a major compacted segment.
+   * @return majorCompacted
+   */
+  public String isMajorCompacted() {
+    return majorCompacted;
+  }
+
+  /**
+   * Set true if it is a major compacted segment.
+   *
+   * @param majorCompacted
+   */
+  public void setMajorCompacted(String majorCompacted) {
+    this.majorCompacted = majorCompacted;
+  }
+
+  /**
+   * To get isDeleted property.
+   *
+   * @return isDeleted
+   */
+  public String getIsDeleted() {
+    return isDeleted;
+  }
+
+  /**
+   * To set isDeleted property.
+   *
+   * @param isDeleted
+   */
+  public void setIsDeleted(String isDeleted) {
+    this.isDeleted = isDeleted;
+  }
+
+  /**
+   * To get the update delta end timestamp
+   *
+   * @return updateDeltaEndTimestamp
+   */
+  public String getUpdateDeltaEndTimestamp() {
+    return updateDeltaEndTimestamp;
+  }
+
+  /**
+   * To set the update delta end timestamp
+   *
+   * @param updateDeltaEndTimestamp
+   */
+  public void setUpdateDeltaEndTimestamp(String updateDeltaEndTimestamp) {
+    this.updateDeltaEndTimestamp = updateDeltaEndTimestamp;
+  }
+
+  /**
+   * To get the update delta start timestamp
+   *
+   * @return updateDeltaStartTimestamp
+   */
+  public String getUpdateDeltaStartTimestamp() {
+    return updateDeltaStartTimestamp;
+  }
+
+  /**
+   * To set the update delta start timestamp
+   *
+   * @param updateDeltaStartTimestamp
+   */
+  public void setUpdateDeltaStartTimestamp(String updateDeltaStartTimestamp) {
+    this.updateDeltaStartTimestamp = updateDeltaStartTimestamp;
+  }
+
+  /**
+   * To get the updateStatusFileName
+   *
+   * @return updateStatusFileName
+   */
+  public String getUpdateStatusFileName() {
+    return updateStatusFileName;
+  }
+
+  /**
+   * To set the updateStatusFileName
+   *
+   * @param updateStatusFileName
+   */
+  public void setUpdateStatusFileName(String updateStatusFileName) {
+    this.updateStatusFileName = updateStatusFileName;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
new file mode 100644
index 0000000..1458d48
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
@@ -0,0 +1,642 @@
+/*
+ * 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.statusmanager;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Arrays;
+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.impl.FileFactory;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperationsImpl;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+import org.apache.carbondata.core.locks.CarbonLockFactory;
+import org.apache.carbondata.core.locks.CarbonLockUtil;
+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.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import com.google.gson.Gson;
+
+/**
+ * Manages Load/Segment status
+ */
+public class SegmentStatusManager {
+
+  private static final LogService LOG =
+      LogServiceFactory.getLogService(SegmentStatusManager.class.getName());
+
+  private AbsoluteTableIdentifier absoluteTableIdentifier;
+
+  public SegmentStatusManager(AbsoluteTableIdentifier absoluteTableIdentifier) {
+    this.absoluteTableIdentifier = absoluteTableIdentifier;
+  }
+
+  /**
+   * This will return the lock object used to lock the table status file before updation.
+   *
+   * @return
+   */
+  public ICarbonLock getTableStatusLock() {
+    return CarbonLockFactory.getCarbonLockObj(absoluteTableIdentifier.getCarbonTableIdentifier(),
+            LockUsage.TABLE_STATUS_LOCK);
+  }
+
+  /**
+   * This method will return last modified time of tablestatus file
+   */
+  public static long getTableStatusLastModifiedTime(AbsoluteTableIdentifier identifier)
+      throws IOException {
+    String tableStatusPath = CarbonStorePath
+        .getCarbonTablePath(identifier.getStorePath(), identifier.getCarbonTableIdentifier())
+        .getTableStatusFilePath();
+    if (!FileFactory.isFileExist(tableStatusPath, FileFactory.getFileType(tableStatusPath))) {
+      return 0L;
+    } else {
+      return FileFactory.getCarbonFile(tableStatusPath, FileFactory.getFileType(tableStatusPath))
+          .getLastModifiedTime();
+    }
+  }
+
+  public List<Long> getUpdateDeltaStartEndTimeStamp(final String loadIds,
+      LoadMetadataDetails[] listOfLoadFolderDetailsArray) {
+    List<Long> updateDeltaStartEndTimestamp = new ArrayList<>();
+    for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
+      if (loadIds.equalsIgnoreCase(loadMetadata.getLoadName())) {
+        // Make sure the Load is not compacted and not marked for delete.
+        if (CarbonCommonConstants.COMPACTED
+                .equalsIgnoreCase(loadMetadata.getLoadStatus())) {
+          return null;
+        } else if (CarbonCommonConstants.MARKED_FOR_DELETE.equals(loadMetadata.getLoadStatus())) {
+          return null;
+        }
+        else {
+          return updateDeltaStartEndTimestamp;
+        }
+      }
+    }
+    return null;
+  }
+
+
+  /**
+   * get valid segment for given table
+   *
+   * @return
+   * @throws IOException
+   */
+  public ValidAndInvalidSegmentsInfo getValidAndInvalidSegments() throws IOException {
+
+    // @TODO: move reading LoadStatus file to separate class
+    List<String> listOfValidSegments = new ArrayList<String>(10);
+    List<String> listOfValidUpdatedSegments = new ArrayList<String>(10);
+    List<String> listOfInvalidSegments = new ArrayList<String>(10);
+    CarbonTablePath carbonTablePath = CarbonStorePath
+            .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+                    absoluteTableIdentifier.getCarbonTableIdentifier());
+    String dataPath = carbonTablePath.getTableStatusFilePath();
+    DataInputStream dataInputStream = null;
+    Gson gsonObjectToRead = new Gson();
+    AtomicFileOperations fileOperation =
+            new AtomicFileOperationsImpl(dataPath, FileFactory.getFileType(dataPath));
+    LoadMetadataDetails[] loadFolderDetailsArray;
+    try {
+      if (FileFactory.isFileExist(dataPath, FileFactory.getFileType(dataPath))) {
+        dataInputStream = fileOperation.openForRead();
+        BufferedReader buffReader =
+                new BufferedReader(new InputStreamReader(dataInputStream, "UTF-8"));
+        loadFolderDetailsArray = gsonObjectToRead.fromJson(buffReader, LoadMetadataDetails[].class);
+        //just directly iterate Array
+        List<LoadMetadataDetails> loadFolderDetails = Arrays.asList(loadFolderDetailsArray);
+        for (LoadMetadataDetails loadMetadataDetails : loadFolderDetails) {
+          if (CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())
+                  || CarbonCommonConstants.MARKED_FOR_UPDATE
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())
+                  || CarbonCommonConstants.STORE_LOADSTATUS_PARTIAL_SUCCESS
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())) {
+            // check for merged loads.
+            if (null != loadMetadataDetails.getMergedLoadName()) {
+              if (!listOfValidSegments.contains(loadMetadataDetails.getMergedLoadName())) {
+                listOfValidSegments.add(loadMetadataDetails.getMergedLoadName());
+              }
+              // if merged load is updated then put it in updated list
+              if (CarbonCommonConstants.MARKED_FOR_UPDATE
+                      .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())) {
+                listOfValidUpdatedSegments.add(loadMetadataDetails.getMergedLoadName());
+              }
+              continue;
+            }
+
+            if (CarbonCommonConstants.MARKED_FOR_UPDATE
+                    .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())) {
+
+              listOfValidUpdatedSegments.add(loadMetadataDetails.getLoadName());
+            }
+            listOfValidSegments.add(loadMetadataDetails.getLoadName());
+          } else if ((CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())
+                  || CarbonCommonConstants.COMPACTED
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus())
+                  || CarbonCommonConstants.MARKED_FOR_DELETE
+                  .equalsIgnoreCase(loadMetadataDetails.getLoadStatus()))) {
+            listOfInvalidSegments.add(loadMetadataDetails.getLoadName());
+          }
+        }
+      }
+    } catch (IOException e) {
+      LOG.error(e);
+      throw e;
+    } finally {
+      try {
+        if (null != dataInputStream) {
+          dataInputStream.close();
+        }
+      } catch (Exception e) {
+        LOG.error(e);
+        throw e;
+      }
+    }
+    return new ValidAndInvalidSegmentsInfo(listOfValidSegments, listOfValidUpdatedSegments,
+            listOfInvalidSegments);
+  }
+
+  /**
+   * This method reads the load metadata file
+   *
+   * @param tableFolderPath
+   * @return
+   */
+  public static LoadMetadataDetails[] readLoadMetadata(String tableFolderPath) {
+    Gson gsonObjectToRead = new Gson();
+    DataInputStream dataInputStream = null;
+    BufferedReader buffReader = null;
+    InputStreamReader inStream = null;
+    String metadataFileName = tableFolderPath + CarbonCommonConstants.FILE_SEPARATOR
+        + CarbonCommonConstants.LOADMETADATA_FILENAME;
+    LoadMetadataDetails[] listOfLoadFolderDetailsArray;
+    AtomicFileOperations fileOperation =
+        new AtomicFileOperationsImpl(metadataFileName, FileFactory.getFileType(metadataFileName));
+
+    try {
+      if (!FileFactory.isFileExist(metadataFileName, FileFactory.getFileType(metadataFileName))) {
+        return new LoadMetadataDetails[0];
+      }
+      dataInputStream = fileOperation.openForRead();
+      inStream = new InputStreamReader(dataInputStream,
+              Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
+      buffReader = new BufferedReader(inStream);
+      listOfLoadFolderDetailsArray =
+          gsonObjectToRead.fromJson(buffReader, LoadMetadataDetails[].class);
+    } catch (IOException e) {
+      return new LoadMetadataDetails[0];
+    } finally {
+      closeStreams(buffReader, inStream, dataInputStream);
+    }
+
+    return listOfLoadFolderDetailsArray;
+  }
+
+  /**
+   * compares two given date strings
+   *
+   * @param loadValue
+   * @param userValue
+   * @return -1 if first arg is less than second arg, 1 if first arg is greater than second arg,
+   * 0 otherwise
+   */
+  private static Integer compareDateValues(Long loadValue, Long userValue) {
+    return loadValue.compareTo(userValue);
+  }
+
+  /**
+   * updates deletion status
+   *
+   * @param loadIds
+   * @param tableFolderPath
+   * @return
+   */
+  public static List<String> updateDeletionStatus(AbsoluteTableIdentifier identifier,
+      List<String> loadIds, String tableFolderPath) throws Exception {
+    CarbonTableIdentifier carbonTableIdentifier = identifier.getCarbonTableIdentifier();
+    ICarbonLock carbonDeleteSegmentLock =
+        CarbonLockFactory.getCarbonLockObj(carbonTableIdentifier, LockUsage.DELETE_SEGMENT_LOCK);
+    ICarbonLock carbonTableStatusLock =
+        CarbonLockFactory.getCarbonLockObj(carbonTableIdentifier, LockUsage.TABLE_STATUS_LOCK);
+    String tableDetails =
+        carbonTableIdentifier.getDatabaseName() + "." + carbonTableIdentifier.getTableName();
+    List<String> invalidLoadIds = new ArrayList<String>(0);
+    try {
+      if (carbonDeleteSegmentLock.lockWithRetries()) {
+        LOG.info("Delete segment lock has been successfully acquired");
+
+        CarbonTablePath carbonTablePath = CarbonStorePath.getCarbonTablePath(
+            identifier.getStorePath(), identifier.getCarbonTableIdentifier());
+        String dataLoadLocation = carbonTablePath.getTableStatusFilePath();
+        LoadMetadataDetails[] listOfLoadFolderDetailsArray = null;
+        if (!FileFactory.isFileExist(dataLoadLocation, FileFactory.getFileType(dataLoadLocation))) {
+          // log error.
+          LOG.error("Load metadata file is not present.");
+          return loadIds;
+        }
+        // read existing metadata details in load metadata.
+        listOfLoadFolderDetailsArray = readLoadMetadata(tableFolderPath);
+        if (listOfLoadFolderDetailsArray != null && listOfLoadFolderDetailsArray.length != 0) {
+          updateDeletionStatus(loadIds, listOfLoadFolderDetailsArray, invalidLoadIds);
+          if (invalidLoadIds.isEmpty()) {
+            // All or None , if anything fails then dont write
+            if(carbonTableStatusLock.lockWithRetries()) {
+              LOG.info("Table status lock has been successfully acquired");
+              // To handle concurrency scenarios, always take latest metadata before writing
+              // into status file.
+              LoadMetadataDetails[] latestLoadMetadataDetails = readLoadMetadata(tableFolderPath);
+              updateLatestTableStatusDetails(listOfLoadFolderDetailsArray,
+                  latestLoadMetadataDetails);
+              writeLoadDetailsIntoFile(dataLoadLocation, listOfLoadFolderDetailsArray);
+            }
+            else {
+              String errorMsg = "Delete segment by id is failed for " + tableDetails
+                  + ". Not able to acquire the table status lock due to other operation running "
+                  + "in the background.";
+              LOG.audit(errorMsg);
+              LOG.error(errorMsg);
+              throw new Exception(errorMsg + " Please try after some time.");
+            }
+
+          } else {
+            return invalidLoadIds;
+          }
+
+        } else {
+          LOG.audit("Delete segment by Id is failed. No matching segment id found.");
+          return loadIds;
+        }
+
+      } else {
+        String errorMsg = "Delete segment by id is failed for " + tableDetails
+            + ". Not able to acquire the delete segment lock due to another delete "
+            + "operation is running in the background.";
+        LOG.audit(errorMsg);
+        LOG.error(errorMsg);
+        throw new Exception(errorMsg + " Please try after some time.");
+      }
+    } catch (IOException e) {
+      LOG.error("IOException" + e.getMessage());
+    } finally {
+      CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.TABLE_STATUS_LOCK);
+      CarbonLockUtil.fileUnlock(carbonDeleteSegmentLock, LockUsage.DELETE_SEGMENT_LOCK);
+    }
+
+    return invalidLoadIds;
+  }
+
+  /**
+   * updates deletion status
+   *
+   * @param loadDate
+   * @param tableFolderPath
+   * @return
+   */
+  public static List<String> updateDeletionStatus(AbsoluteTableIdentifier identifier,
+      String loadDate, String tableFolderPath, Long loadStartTime) throws Exception {
+    CarbonTableIdentifier carbonTableIdentifier = identifier.getCarbonTableIdentifier();
+    ICarbonLock carbonDeleteSegmentLock =
+        CarbonLockFactory.getCarbonLockObj(carbonTableIdentifier, LockUsage.DELETE_SEGMENT_LOCK);
+    ICarbonLock carbonTableStatusLock =
+        CarbonLockFactory.getCarbonLockObj(carbonTableIdentifier, LockUsage.TABLE_STATUS_LOCK);
+    String tableDetails =
+        carbonTableIdentifier.getDatabaseName() + "." + carbonTableIdentifier.getTableName();
+    List<String> invalidLoadTimestamps = new ArrayList<String>(0);
+    try {
+      if (carbonDeleteSegmentLock.lockWithRetries()) {
+        LOG.info("Delete segment lock has been successfully acquired");
+
+        CarbonTablePath carbonTablePath = CarbonStorePath.getCarbonTablePath(
+            identifier.getStorePath(), identifier.getCarbonTableIdentifier());
+        String dataLoadLocation = carbonTablePath.getTableStatusFilePath();
+        LoadMetadataDetails[] listOfLoadFolderDetailsArray = null;
+
+        if (!FileFactory.isFileExist(dataLoadLocation, FileFactory.getFileType(dataLoadLocation))) {
+          // log error.
+          LOG.error("Error message: " + "Load metadata file is not present.");
+          invalidLoadTimestamps.add(loadDate);
+          return invalidLoadTimestamps;
+        }
+        // read existing metadata details in load metadata.
+        listOfLoadFolderDetailsArray = readLoadMetadata(tableFolderPath);
+        if (listOfLoadFolderDetailsArray != null && listOfLoadFolderDetailsArray.length != 0) {
+          updateDeletionStatus(loadDate, listOfLoadFolderDetailsArray, invalidLoadTimestamps,
+              loadStartTime);
+          if (invalidLoadTimestamps.isEmpty()) {
+            if(carbonTableStatusLock.lockWithRetries()) {
+              LOG.info("Table status lock has been successfully acquired.");
+              // To handle concurrency scenarios, always take latest metadata before writing
+              // into status file.
+              LoadMetadataDetails[] latestLoadMetadataDetails = readLoadMetadata(tableFolderPath);
+              updateLatestTableStatusDetails(listOfLoadFolderDetailsArray,
+                  latestLoadMetadataDetails);
+              writeLoadDetailsIntoFile(dataLoadLocation, listOfLoadFolderDetailsArray);
+            }
+            else {
+
+              String errorMsg = "Delete segment by date is failed for " + tableDetails
+                  + ". Not able to acquire the table status lock due to other operation running "
+                  + "in the background.";
+              LOG.audit(errorMsg);
+              LOG.error(errorMsg);
+              throw new Exception(errorMsg + " Please try after some time.");
+
+            }
+          } else {
+            return invalidLoadTimestamps;
+          }
+
+        } else {
+          LOG.audit("Delete segment by date is failed. No matching segment found.");
+          invalidLoadTimestamps.add(loadDate);
+          return invalidLoadTimestamps;
+        }
+
+      } else {
+        String errorMsg = "Delete segment by date is failed for " + tableDetails
+            + ". Not able to acquire the delete segment lock due to another delete "
+            + "operation is running in the background.";
+        LOG.audit(errorMsg);
+        LOG.error(errorMsg);
+        throw new Exception(errorMsg + " Please try after some time.");
+      }
+    } catch (IOException e) {
+      LOG.error("Error message: " + "IOException" + e.getMessage());
+    } finally {
+      CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.TABLE_STATUS_LOCK);
+      CarbonLockUtil.fileUnlock(carbonDeleteSegmentLock, LockUsage.DELETE_SEGMENT_LOCK);
+    }
+
+    return invalidLoadTimestamps;
+  }
+
+  /**
+   * writes load details into a given file at @param dataLoadLocation
+   *
+   * @param dataLoadLocation
+   * @param listOfLoadFolderDetailsArray
+   * @throws IOException
+   */
+  public static void writeLoadDetailsIntoFile(String dataLoadLocation,
+      LoadMetadataDetails[] listOfLoadFolderDetailsArray) throws IOException {
+    AtomicFileOperations fileWrite =
+        new AtomicFileOperationsImpl(dataLoadLocation, FileFactory.getFileType(dataLoadLocation));
+    BufferedWriter brWriter = null;
+    DataOutputStream dataOutputStream = null;
+    Gson gsonObjectToWrite = new Gson();
+    // write the updated data into the metadata file.
+
+    try {
+      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
+      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
+              Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
+
+      String metadataInstance = gsonObjectToWrite.toJson(listOfLoadFolderDetailsArray);
+      brWriter.write(metadataInstance);
+    } catch (IOException ioe) {
+      LOG.error("Error message: " + ioe.getLocalizedMessage());
+    } finally {
+      if (null != brWriter) {
+        brWriter.flush();
+      }
+      CarbonUtil.closeStreams(brWriter);
+      fileWrite.close();
+    }
+
+  }
+
+  /**
+   * updates deletion status details for each load and returns invalidLoadIds
+   *
+   * @param loadIds
+   * @param listOfLoadFolderDetailsArray
+   * @param invalidLoadIds
+   * @return invalidLoadIds
+   */
+  private static List<String> updateDeletionStatus(List<String> loadIds,
+      LoadMetadataDetails[] listOfLoadFolderDetailsArray, List<String> invalidLoadIds) {
+    for (String loadId : loadIds) {
+      boolean loadFound = false;
+      // For each load id loop through data and if the
+      // load id is found then mark
+      // the metadata as deleted.
+      for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
+
+        if (loadId.equalsIgnoreCase(loadMetadata.getLoadName())) {
+          // if the segment is compacted then no need to delete that.
+          if (CarbonCommonConstants.COMPACTED
+                  .equalsIgnoreCase(loadMetadata.getLoadStatus())) {
+            LOG.error("Cannot delete the Segment which is compacted. Segment is " + loadId);
+            invalidLoadIds.add(loadId);
+            return invalidLoadIds;
+          }
+          if (!CarbonCommonConstants.MARKED_FOR_DELETE.equals(loadMetadata.getLoadStatus())) {
+            loadFound = true;
+            loadMetadata.setLoadStatus(CarbonCommonConstants.MARKED_FOR_DELETE);
+            loadMetadata.setModificationOrdeletionTimesStamp(CarbonUpdateUtil.readCurrentTime());
+            LOG.info("Segment ID " + loadId + " Marked for Delete");
+          }
+          break;
+        }
+      }
+
+      if (!loadFound) {
+        LOG.audit("Delete segment by ID is failed. No matching segment id found :" + loadId);
+        invalidLoadIds.add(loadId);
+        return invalidLoadIds;
+      }
+    }
+    return invalidLoadIds;
+  }
+
+  /**
+   * updates deletion status details for load and returns invalidLoadTimestamps
+   *
+   * @param loadDate
+   * @param listOfLoadFolderDetailsArray
+   * @param invalidLoadTimestamps
+   * @return invalidLoadTimestamps
+   */
+  public static List<String> updateDeletionStatus(String loadDate,
+      LoadMetadataDetails[] listOfLoadFolderDetailsArray, List<String> invalidLoadTimestamps,
+      Long loadStartTime) {
+    // For each load timestamp loop through data and if the
+    // required load timestamp is found then mark
+    // the metadata as deleted.
+    boolean loadFound = false;
+    String loadStartTimeString = "Load Start Time: ";
+    for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
+      Integer result = compareDateValues(loadMetadata.getLoadStartTimeAsLong(), loadStartTime);
+      if (result < 0) {
+        if (CarbonCommonConstants.COMPACTED
+            .equalsIgnoreCase(loadMetadata.getLoadStatus())) {
+          LOG.info("Ignoring the segment : " + loadMetadata.getLoadName()
+              + "as the segment has been compacted.");
+          continue;
+        }
+        if (!CarbonCommonConstants.MARKED_FOR_DELETE.equals(loadMetadata.getLoadStatus())) {
+          loadFound = true;
+          updateSegmentMetadataDetails(loadMetadata);
+          LOG.info("Info: " +
+              loadStartTimeString + loadMetadata.getLoadStartTime() +
+              " Marked for Delete");
+        }
+      }
+
+    }
+
+    if (!loadFound) {
+      invalidLoadTimestamps.add(loadDate);
+      LOG.audit("Delete segment by date is failed. No matching segment found.");
+      return invalidLoadTimestamps;
+    }
+    return invalidLoadTimestamps;
+  }
+
+  /**
+   * This method closes the streams
+   *
+   * @param streams - streams to close.
+   */
+  private static void closeStreams(Closeable... streams) {
+    // Added if to avoid NullPointerException in case one stream is being passed as null
+    if (null != streams) {
+      for (Closeable stream : streams) {
+        if (null != stream) {
+          try {
+            stream.close();
+          } catch (IOException e) {
+            LOG.error("Error while closing stream" + stream);
+          }
+        }
+      }
+    }
+  }
+
+  /**
+   * updates table status details using latest metadata
+   *
+   * @param oldMetadata
+   * @param newMetadata
+   * @return
+   */
+
+  public static List<LoadMetadataDetails> updateLatestTableStatusDetails(
+      LoadMetadataDetails[] oldMetadata, LoadMetadataDetails[] newMetadata) {
+
+    List<LoadMetadataDetails> newListMetadata =
+        new ArrayList<LoadMetadataDetails>(Arrays.asList(newMetadata));
+    for (LoadMetadataDetails oldSegment : oldMetadata) {
+      if (CarbonCommonConstants.MARKED_FOR_DELETE.equalsIgnoreCase(oldSegment.getLoadStatus())) {
+        updateSegmentMetadataDetails(newListMetadata.get(newListMetadata.indexOf(oldSegment)));
+      }
+    }
+    return newListMetadata;
+  }
+
+  /**
+   * updates segment status and modificaton time details
+   *
+   * @param loadMetadata
+   */
+  public static void updateSegmentMetadataDetails(LoadMetadataDetails loadMetadata) {
+    // update status only if the segment is not marked for delete
+    if (!CarbonCommonConstants.MARKED_FOR_DELETE.equalsIgnoreCase(loadMetadata.getLoadStatus())) {
+      loadMetadata.setLoadStatus(CarbonCommonConstants.MARKED_FOR_DELETE);
+      loadMetadata.setModificationOrdeletionTimesStamp(CarbonUpdateUtil.readCurrentTime());
+    }
+  }
+
+  /**
+   * This API will return the update status file name.
+   * @param segmentList
+   * @return
+   */
+  public String getUpdateStatusFileName(LoadMetadataDetails[] segmentList) {
+    if(segmentList.length == 0) {
+      return "";
+    }
+    else {
+      for(LoadMetadataDetails eachSeg : segmentList) {
+        // file name stored in 0th segment.
+        if (eachSeg.getLoadName().equalsIgnoreCase("0")) {
+          return eachSeg.getUpdateStatusFileName();
+        }
+      }
+    }
+    return "";
+  }
+
+  /**
+   * getting the task numbers present in the segment.
+   * @param segmentId
+   * @return
+   */
+  public List<String> getUpdatedTasksDetailsForSegment(String segmentId, SegmentUpdateStatusManager
+          updateStatusManager) {
+    List<String> taskList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    List<String> list = updateStatusManager.getUpdateDeltaFiles(segmentId);
+    for (String eachFileName : list) {
+      taskList.add(CarbonTablePath.DataFileUtil.getTaskNo(eachFileName));
+    }
+    return taskList;
+  }
+
+
+  public static class ValidAndInvalidSegmentsInfo {
+    private final List<String> listOfValidSegments;
+    private final List<String> listOfValidUpdatedSegments;
+    private final List<String> listOfInvalidSegments;
+
+    private ValidAndInvalidSegmentsInfo(List<String> listOfValidSegments,
+        List<String> listOfValidUpdatedSegments, List<String> listOfInvalidUpdatedSegments) {
+      this.listOfValidSegments = listOfValidSegments;
+      this.listOfValidUpdatedSegments = listOfValidUpdatedSegments;
+      this.listOfInvalidSegments = listOfInvalidUpdatedSegments;
+    }
+    public List<String> getInvalidSegments() {
+      return listOfInvalidSegments;
+    }
+    public List<String> getValidSegments() {
+      return listOfValidSegments;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
new file mode 100644
index 0000000..f14485b
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
@@ -0,0 +1,990 @@
+/*
+ * 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.statusmanager;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+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.fileoperations.AtomicFileOperations;
+import org.apache.carbondata.core.fileoperations.AtomicFileOperationsImpl;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+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.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.mutate.SegmentUpdateDetails;
+import org.apache.carbondata.core.mutate.TupleIdEnum;
+import org.apache.carbondata.core.mutate.UpdateVO;
+import org.apache.carbondata.core.reader.CarbonDeleteFilesDataReader;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonStorePath;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import com.google.gson.Gson;
+
+/**
+ * Manages Segment & block status of carbon table for Delete operation
+ */
+public class SegmentUpdateStatusManager {
+
+  /**
+   * logger
+   */
+  private static final LogService LOG =
+      LogServiceFactory.getLogService(SegmentUpdateStatusManager.class.getName());
+
+  private AbsoluteTableIdentifier absoluteTableIdentifier;
+  private LoadMetadataDetails[] segmentDetails;
+  private SegmentUpdateDetails[] updateDetails;
+  private CarbonTablePath carbonTablePath;
+  private Map<String, SegmentUpdateDetails> blockAndDetailsMap;
+
+  /**
+   * @param absoluteTableIdentifier
+   */
+  public SegmentUpdateStatusManager(AbsoluteTableIdentifier absoluteTableIdentifier) {
+    this.absoluteTableIdentifier = absoluteTableIdentifier;
+    carbonTablePath = CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+        absoluteTableIdentifier.getCarbonTableIdentifier());
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+    // current it is used only for read function scenarios, as file update always requires to work
+    // on latest file status.
+    segmentDetails =
+        segmentStatusManager.readLoadMetadata(carbonTablePath.getMetadataDirectoryPath());
+    updateDetails = readLoadMetadata();
+    populateMap();
+  }
+
+  /**
+   * populate the block and its details in a map.
+   */
+  private void populateMap() {
+    blockAndDetailsMap = new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    for(SegmentUpdateDetails blockDetails : updateDetails) {
+
+      String blockIdentifier = CarbonUpdateUtil
+          .getSegmentBlockNameKey(blockDetails.getSegmentName(), blockDetails.getActualBlockName());
+
+      blockAndDetailsMap.put(blockIdentifier, blockDetails);
+
+    }
+
+  }
+
+  /**
+   *
+   * @param segID
+   * @param actualBlockName
+   * @return null if block is not present in segment update status.
+   */
+  public SegmentUpdateDetails getDetailsForABlock(String segID, String actualBlockName) {
+
+    String blockIdentifier = CarbonUpdateUtil
+        .getSegmentBlockNameKey(segID, actualBlockName);
+
+    return blockAndDetailsMap.get(blockIdentifier);
+
+  }
+
+  /**
+   *
+   * @param key will be like (segid/blockname)  0/0-0-5464654654654
+   * @return
+   */
+  public SegmentUpdateDetails getDetailsForABlock(String key) {
+
+    return blockAndDetailsMap.get(key);
+
+  }
+
+
+
+  /**
+   * Returns the LoadMetadata Details
+   * @return
+   */
+  public LoadMetadataDetails[] getLoadMetadataDetails() {
+    return segmentDetails;
+  }
+
+  /**
+   *
+   * @param loadMetadataDetails
+   */
+  public void setLoadMetadataDetails(LoadMetadataDetails[] loadMetadataDetails) {
+    this.segmentDetails = loadMetadataDetails;
+  }
+
+  /**
+   * Returns the UpdateStatus Details.
+   * @return
+   */
+  public SegmentUpdateDetails[] getUpdateStatusDetails() {
+    return updateDetails;
+  }
+
+  /**
+   *
+   * @param segmentUpdateDetails
+   */
+  public void setUpdateStatusDetails(SegmentUpdateDetails[] segmentUpdateDetails) {
+    this.updateDetails = segmentUpdateDetails;
+  }
+
+  /**
+   * This will return the lock object used to lock the table update status file before updation.
+   *
+   * @return
+   */
+  public ICarbonLock getTableUpdateStatusLock() {
+    return CarbonLockFactory.getCarbonLockObj(absoluteTableIdentifier.getCarbonTableIdentifier(),
+        LockUsage.TABLE_UPDATE_STATUS_LOCK);
+  }
+
+  /**
+   * Returns all delete delta files of specified block
+   *
+   * @param tupleId
+   * @return
+   * @throws Exception
+   */
+  public List<String> getDeleteDeltaFiles(String tupleId) throws Exception {
+    return getDeltaFiles(tupleId, CarbonCommonConstants.DELETE_DELTA_FILE_EXT);
+  }
+
+
+  /**
+   * Returns all update delta files of specified Segment.
+   *
+   * @param segmentId
+   * @return
+   * @throws Exception
+   */
+  public List<String> getUpdateDeltaFiles(final String segmentId) {
+    List<String> updatedDeltaFilesList =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    String endTimeStamp = "";
+    String startTimeStamp = "";
+    String segmentPath = carbonTablePath.getCarbonDataDirectoryPath("0", segmentId);
+    CarbonFile segDir =
+        FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+    for (LoadMetadataDetails eachSeg : segmentDetails) {
+      if (eachSeg.getLoadName().equalsIgnoreCase(segmentId)) {
+        // if the segment is found then take the start and end time stamp.
+        startTimeStamp = eachSeg.getUpdateDeltaStartTimestamp();
+        endTimeStamp = eachSeg.getUpdateDeltaEndTimestamp();
+      }
+    }
+    // if start timestamp is empty then no update delta is found. so return empty list.
+    if (startTimeStamp.isEmpty()) {
+      return updatedDeltaFilesList;
+    }
+    final Long endTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(endTimeStamp);
+    final Long startTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(startTimeStamp);
+
+    // else scan the segment for the delta files with the respective timestamp.
+    CarbonFile[] files = segDir.listFiles(new CarbonFileFilter() {
+
+      @Override public boolean accept(CarbonFile pathName) {
+        String fileName = pathName.getName();
+        if (fileName.endsWith(CarbonCommonConstants.UPDATE_DELTA_FILE_EXT)) {
+          String firstPart = fileName.substring(0, fileName.indexOf('.'));
+
+          long timestamp = Long.parseLong(firstPart
+              .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
+                  firstPart.length()));
+          if (Long.compare(timestamp, endTimeStampFinal) <= 0
+              && Long.compare(timestamp, startTimeStampFinal) >= 0) {
+
+            // if marked for delete then it is invalid.
+            if (!isBlockValid(segmentId, fileName)) {
+              return false;
+            }
+
+            return true;
+          }
+        }
+        return false;
+      }
+    });
+
+    for (CarbonFile cfile : files) {
+      updatedDeltaFilesList.add(cfile.getCanonicalPath());
+    }
+
+    return updatedDeltaFilesList;
+  }
+
+  /**
+   * Returns all deleted records of specified block
+   *
+   * @param tupleId
+   * @return
+   * @throws Exception
+   */
+  public int[] getDeleteDeltaDataFromAllFiles(String tupleId) throws Exception {
+    List<String> deltaFiles = getDeltaFiles(tupleId, CarbonCommonConstants.DELETE_DELTA_FILE_EXT);
+    CarbonDeleteFilesDataReader dataReader = new CarbonDeleteFilesDataReader();
+    String blockletId = CarbonUpdateUtil.getRequiredFieldFromTID(tupleId, TupleIdEnum.BLOCKLET_ID);
+    return dataReader.getDeleteDataFromAllFiles(deltaFiles, blockletId);
+  }
+
+
+
+  /**
+   * Returns all delta file paths of specified block
+   *
+   * @param tupleId
+   * @param extension
+   * @return
+   * @throws Exception
+   */
+  public List<String> getDeltaFiles(String tupleId, String extension) throws Exception {
+    try {
+      CarbonTablePath carbonTablePath = CarbonStorePath
+          .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+              absoluteTableIdentifier.getCarbonTableIdentifier());
+      String segment = CarbonUpdateUtil.getRequiredFieldFromTID(tupleId, TupleIdEnum.SEGMENT_ID);
+      String carbonDataDirectoryPath = carbonTablePath.getCarbonDataDirectoryPath("0", segment);
+      String completeBlockName = CarbonTablePath.addDataPartPrefix(
+          CarbonUpdateUtil.getRequiredFieldFromTID(tupleId, TupleIdEnum.BLOCK_ID)
+              + CarbonCommonConstants.FACT_FILE_EXT);
+      String blockPath =
+          carbonDataDirectoryPath + CarbonCommonConstants.FILE_SEPARATOR + completeBlockName;
+      CarbonFile file = FileFactory.getCarbonFile(blockPath, FileFactory.getFileType(blockPath));
+      if (!file.exists()) {
+        throw new Exception("Invalid tuple id " + tupleId);
+      }
+      String blockNameWithoutExtn = completeBlockName.substring(0, completeBlockName.indexOf('.'));
+      //blockName without timestamp
+      final String blockNameFromTuple =
+          blockNameWithoutExtn.substring(0, blockNameWithoutExtn.lastIndexOf("-"));
+      SegmentUpdateDetails[] listOfSegmentUpdateDetailsArray =
+          readLoadMetadata();
+      return getDeltaFiles(file, blockNameFromTuple, listOfSegmentUpdateDetailsArray, extension,
+          segment);
+
+    } catch (Exception ex) {
+      String errorMsg = "Invalid tuple id " + tupleId;
+      LOG.error(errorMsg);
+      throw new Exception(errorMsg);
+    }
+  }
+
+  /**
+   * This method returns the list of Blocks associated with the segment
+   * from the SegmentUpdateDetails List.
+   * @param segmentName
+   * @return
+   */
+  public List<String> getBlockNameFromSegment(String segmentName) {
+    List<String> blockNames = new ArrayList<String>();
+    for (SegmentUpdateDetails block : updateDetails) {
+      if (block.getSegmentName().equalsIgnoreCase(segmentName) && !CarbonUpdateUtil
+          .isBlockInvalid(block.getStatus())) {
+        blockNames.add(block.getBlockName());
+      }
+    }
+    return blockNames;
+  }
+
+  /**
+   *
+   * @param segName
+   * @param blockName
+   * @return
+   */
+  public boolean isBlockValid(String segName, String blockName) {
+
+    SegmentUpdateDetails details = getDetailsForABlock(segName, blockName);
+
+    if (details == null || !CarbonUpdateUtil.isBlockInvalid(details.getStatus())) {
+      return true;
+    }
+
+    return false;
+  }
+  /**
+   * Returns all delta file paths of specified block
+   *
+   * @param blockDir
+   * @param blockNameFromTuple
+   * @param listOfSegmentUpdateDetailsArray
+   * @param extension
+   * @return
+   */
+  public List<String> getDeltaFiles(CarbonFile blockDir, final String blockNameFromTuple,
+      SegmentUpdateDetails[] listOfSegmentUpdateDetailsArray,
+      final String extension,
+      String segment) {
+    List<String> deleteFileList = null;
+    for (SegmentUpdateDetails block : listOfSegmentUpdateDetailsArray) {
+      if (block.getBlockName().equalsIgnoreCase(blockNameFromTuple) && block.getSegmentName()
+          .equalsIgnoreCase(segment) && !CarbonUpdateUtil.isBlockInvalid(block.getStatus())) {
+        final long deltaStartTimestamp = getStartTimeOfDeltaFile(extension, block);
+        // If there is no delete delete file , then return null
+        if(deltaStartTimestamp == 0) {
+          return deleteFileList;
+        }
+        final long deltaEndTimeStamp = getEndTimeOfDeltaFile(extension, block);
+
+        // final long deltaEndTimeStamp = block.getDeleteDeltaEndTimeAsLong();
+        // final long deltaStartTimestamp = block.getDeleteDeltaStartTimeAsLong();
+        return getFilePaths(blockDir, blockNameFromTuple, extension, deleteFileList,
+            deltaStartTimestamp, deltaEndTimeStamp);
+      }
+    }
+    return deleteFileList;
+  }
+
+  private List<String> getFilePaths(CarbonFile blockDir, final String blockNameFromTuple,
+      final String extension, List<String> deleteFileList, final long deltaStartTimestamp,
+      final long deltaEndTimeStamp) {
+    CarbonFile[] files = blockDir.getParentFile().listFiles(new CarbonFileFilter() {
+
+      @Override public boolean accept(CarbonFile pathName) {
+        String fileName = pathName.getName();
+        if (fileName.endsWith(extension)) {
+          String firstPart = fileName.substring(0, fileName.indexOf('.'));
+          String blockName =
+              firstPart.substring(0, firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN));
+          long timestamp = Long.parseLong(firstPart
+              .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
+                  firstPart.length()));
+          if (blockNameFromTuple.equals(blockName) && ((
+              Long.compare(timestamp, deltaEndTimeStamp ) <= 0) && (
+              Long.compare(timestamp, deltaStartTimestamp) >= 0))) {
+            return true;
+          }
+        }
+        return false;
+      }
+    });
+
+    for (CarbonFile cfile : files) {
+      if(null == deleteFileList) {
+        deleteFileList = new ArrayList<String>(files.length);
+      }
+      deleteFileList.add(cfile.getCanonicalPath());
+    }
+    return deleteFileList;
+  }
+
+  /**
+   * Return all delta file for a block.
+   * @param segmentId
+   * @param blockName
+   * @return
+   */
+  public CarbonFile[] getDeleteDeltaFilesList(final String segmentId, final String blockName) {
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    String segmentPath = carbonTablePath.getCarbonDataDirectoryPath("0", segmentId);
+
+    CarbonFile segDir =
+        FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+
+    for (SegmentUpdateDetails block : updateDetails) {
+      if ((block.getBlockName().equalsIgnoreCase(blockName)) &&
+          (block.getSegmentName().equalsIgnoreCase(segmentId))
+          && !CarbonUpdateUtil.isBlockInvalid((block.getStatus()))) {
+        final long deltaStartTimestamp =
+            getStartTimeOfDeltaFile(CarbonCommonConstants.DELETE_DELTA_FILE_EXT, block);
+        final long deltaEndTimeStamp =
+            getEndTimeOfDeltaFile(CarbonCommonConstants.DELETE_DELTA_FILE_EXT, block);
+        CarbonFile[] files = segDir.listFiles(new CarbonFileFilter() {
+
+          @Override public boolean accept(CarbonFile pathName) {
+            String fileName = pathName.getName();
+            if (fileName.endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT)) {
+              String firstPart = fileName.substring(0, fileName.indexOf('.'));
+              String blkName = firstPart.substring(0, firstPart.lastIndexOf("-"));
+              long timestamp = Long.parseLong(
+                  firstPart.substring(firstPart.lastIndexOf("-") + 1, firstPart.length()));
+              if (blockName.equals(blkName) && (Long.compare(timestamp, deltaEndTimeStamp) <= 0)
+                  && (Long.compare(timestamp, deltaStartTimestamp) >= 0)) {
+                return true;
+              }
+            }
+            return false;
+          }
+        });
+
+        return files;
+      }
+    }
+    return null;
+  }
+
+  /**
+   * Returns all update delta files of specified Segment.
+   *
+   * @param segmentId
+   * @param validUpdateFiles if true then only the valid range files will be returned.
+   * @return
+   */
+  public CarbonFile[] getUpdateDeltaFilesList(String segmentId, final boolean validUpdateFiles,
+      final String fileExtension, final boolean excludeOriginalFact,
+      CarbonFile[] allFilesOfSegment) {
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+    String endTimeStamp = "";
+    String startTimeStamp = "";
+    long factTimeStamp = 0;
+
+    LoadMetadataDetails[] segmentDetails =
+        segmentStatusManager.readLoadMetadata(carbonTablePath.getMetadataDirectoryPath());
+
+    for (LoadMetadataDetails eachSeg : segmentDetails) {
+      if (eachSeg.getLoadName().equalsIgnoreCase(segmentId)) {
+        // if the segment is found then take the start and end time stamp.
+        startTimeStamp = eachSeg.getUpdateDeltaStartTimestamp();
+        endTimeStamp = eachSeg.getUpdateDeltaEndTimestamp();
+        factTimeStamp = eachSeg.getLoadStartTime();
+      }
+    }
+
+    // if start timestamp is empty then no update delta is found. so return empty list.
+    if (startTimeStamp.isEmpty()) {
+      return new CarbonFile[0];
+    }
+
+    final Long endTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(endTimeStamp);
+    final Long startTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(startTimeStamp);
+    final long factTimeStampFinal = factTimeStamp;
+
+    List<CarbonFile> listOfCarbonFiles =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    // else scan the segment for the delta files with the respective timestamp.
+
+    for (CarbonFile eachFile : allFilesOfSegment) {
+
+      String fileName = eachFile.getName();
+      if (fileName.endsWith(fileExtension)) {
+        String firstPart = fileName.substring(0, fileName.indexOf('.'));
+
+        long timestamp = Long.parseLong(firstPart
+            .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
+                firstPart.length()));
+
+        if (excludeOriginalFact) {
+          if (Long.compare(factTimeStampFinal, timestamp) == 0) {
+            continue;
+          }
+        }
+
+        if (validUpdateFiles) {
+          if (Long.compare(timestamp, endTimeStampFinal) <= 0
+              && Long.compare(timestamp, startTimeStampFinal) >= 0) {
+            listOfCarbonFiles.add(eachFile);
+          }
+        } else {
+          // invalid cases.
+          if (Long.compare(timestamp, startTimeStampFinal) < 0) {
+            listOfCarbonFiles.add(eachFile);
+          }
+        }
+      }
+    }
+
+    return listOfCarbonFiles.toArray(new CarbonFile[listOfCarbonFiles.size()]);
+  }
+
+  /**
+   * Returns all update delta files of specified Segment.
+   *
+   * @param segmentId
+   * @param validUpdateFiles
+   * @param fileExtension
+   * @param excludeOriginalFact
+   * @param allFilesOfSegment
+   * @return
+   */
+  public CarbonFile[] getUpdateDeltaFilesForSegment(String segmentId,
+      final boolean validUpdateFiles, final String fileExtension, final boolean excludeOriginalFact,
+      CarbonFile[] allFilesOfSegment) {
+
+    String endTimeStamp = "";
+    String startTimeStamp = "";
+    long factTimeStamp = 0;
+
+    for (LoadMetadataDetails eachSeg : segmentDetails) {
+      if (eachSeg.getLoadName().equalsIgnoreCase(segmentId)) {
+        // if the segment is found then take the start and end time stamp.
+        startTimeStamp = eachSeg.getUpdateDeltaStartTimestamp();
+        endTimeStamp = eachSeg.getUpdateDeltaEndTimestamp();
+        factTimeStamp = eachSeg.getLoadStartTime();
+      }
+    }
+
+    // if start timestamp is empty then no update delta is found. so return empty list.
+    if (startTimeStamp.isEmpty()) {
+      return new CarbonFile[0];
+    }
+
+    final Long endTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(endTimeStamp);
+    final Long startTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(startTimeStamp);
+    final long factTimeStampFinal = factTimeStamp;
+
+    List<CarbonFile> listOfCarbonFiles =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    // else scan the segment for the delta files with the respective timestamp.
+
+    for (CarbonFile eachFile : allFilesOfSegment) {
+
+      String fileName = eachFile.getName();
+      if (fileName.endsWith(fileExtension)) {
+        String firstPart = fileName.substring(0, fileName.indexOf('.'));
+
+        long timestamp = Long.parseLong(firstPart
+            .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
+                firstPart.length()));
+
+        if (excludeOriginalFact) {
+          if (Long.compare(factTimeStampFinal, timestamp) == 0) {
+            continue;
+          }
+        }
+
+        if (validUpdateFiles) {
+          if (Long.compare(timestamp, endTimeStampFinal) <= 0
+              && Long.compare(timestamp, startTimeStampFinal) >= 0) {
+
+            boolean validBlock = true;
+
+            for(SegmentUpdateDetails blockDetails : getUpdateStatusDetails()) {
+              if (blockDetails.getActualBlockName().equalsIgnoreCase(eachFile.getName())
+                  && CarbonUpdateUtil.isBlockInvalid(blockDetails.getStatus())) {
+                validBlock = false;
+              }
+            }
+
+            if (validBlock) {
+              listOfCarbonFiles.add(eachFile);
+            }
+
+          }
+        } else {
+          // invalid cases.
+          if (Long.compare(timestamp, startTimeStampFinal) < 0) {
+            listOfCarbonFiles.add(eachFile);
+          }
+        }
+      }
+    }
+
+    return listOfCarbonFiles.toArray(new CarbonFile[listOfCarbonFiles.size()]);
+  }
+
+  /**
+   *
+   * @param extension
+   * @param block
+   * @return
+   */
+  private long getStartTimeOfDeltaFile(String extension, SegmentUpdateDetails block) {
+    long startTimestamp;
+    switch (extension) {
+      case CarbonCommonConstants.DELETE_DELTA_FILE_EXT:
+        startTimestamp = block.getDeleteDeltaStartTimeAsLong();
+        break;
+      default:
+        startTimestamp = 0;
+    }
+    return startTimestamp;
+  }
+
+  /**
+   *
+   * @param extension
+   * @param block
+   * @return
+   */
+  private long getEndTimeOfDeltaFile(String extension, SegmentUpdateDetails block) {
+    long endTimestamp;
+    switch (extension) {
+      case CarbonCommonConstants.DELETE_DELTA_FILE_EXT:
+        endTimestamp = block.getDeleteDeltaEndTimeAsLong();
+        break;
+      default: endTimestamp = 0;
+    }
+    return endTimestamp;
+  }
+
+
+  /**
+   * This method loads segment update details
+   *
+   * @return
+   */
+  public SegmentUpdateDetails[] readLoadMetadata() {
+    Gson gsonObjectToRead = new Gson();
+    DataInputStream dataInputStream = null;
+    BufferedReader buffReader = null;
+    InputStreamReader inStream = null;
+    SegmentUpdateDetails[] listOfSegmentUpdateDetailsArray;
+
+    // get the updated status file identifier from the table status.
+    String tableUpdateStatusIdentifier = getUpdatedStatusIdentifier();
+
+    if(null == tableUpdateStatusIdentifier) {
+      return new SegmentUpdateDetails[0];
+    }
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    String tableUpdateStatusPath =
+        carbonTablePath.getMetadataDirectoryPath() + CarbonCommonConstants.FILE_SEPARATOR
+            + tableUpdateStatusIdentifier;
+    AtomicFileOperations fileOperation = new AtomicFileOperationsImpl(tableUpdateStatusPath,
+        FileFactory.getFileType(tableUpdateStatusPath));
+
+    try {
+      if (!FileFactory
+          .isFileExist(tableUpdateStatusPath, FileFactory.getFileType(tableUpdateStatusPath))) {
+        return new SegmentUpdateDetails[0];
+      }
+      dataInputStream = fileOperation.openForRead();
+      inStream = new InputStreamReader(dataInputStream,
+          CarbonCommonConstants.CARBON_DEFAULT_STREAM_ENCODEFORMAT);
+      buffReader = new BufferedReader(inStream);
+      listOfSegmentUpdateDetailsArray =
+          gsonObjectToRead.fromJson(buffReader, SegmentUpdateDetails[].class);
+    } catch (IOException e) {
+      return new SegmentUpdateDetails[0];
+    } finally {
+      closeStreams(buffReader, inStream, dataInputStream);
+    }
+
+    return listOfSegmentUpdateDetailsArray;
+  }
+
+  /**
+   * @return updateStatusFileName
+   */
+  private String getUpdatedStatusIdentifier() {
+    SegmentStatusManager ssm = new SegmentStatusManager(absoluteTableIdentifier);
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+    LoadMetadataDetails[] loadDetails =
+        ssm.readLoadMetadata(carbonTablePath.getMetadataDirectoryPath());
+    if (loadDetails.length == 0) {
+      return null;
+    }
+    return loadDetails[0].getUpdateStatusFileName();
+  }
+
+  /**
+   * writes segment update details into a given file at @param dataLoadLocation
+   *
+   * @param listOfSegmentUpdateDetailsArray
+   * @throws IOException
+   */
+  public void writeLoadDetailsIntoFile(List<SegmentUpdateDetails> listOfSegmentUpdateDetailsArray,
+      String updateStatusFileIdentifier) throws IOException {
+
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+
+    String fileLocation =
+        carbonTablePath.getMetadataDirectoryPath() + CarbonCommonConstants.FILE_SEPARATOR
+            + CarbonUpdateUtil.getUpdateStatusFileName(updateStatusFileIdentifier);
+
+    AtomicFileOperations fileWrite =
+        new AtomicFileOperationsImpl(fileLocation, FileFactory.getFileType(fileLocation));
+    BufferedWriter brWriter = null;
+    DataOutputStream dataOutputStream = null;
+    Gson gsonObjectToWrite = new Gson();
+    // write the updated data into the metadata file.
+
+    try {
+      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
+      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
+          CarbonCommonConstants.CARBON_DEFAULT_STREAM_ENCODEFORMAT));
+
+      String metadataInstance = gsonObjectToWrite.toJson(listOfSegmentUpdateDetailsArray);
+      brWriter.write(metadataInstance);
+    } catch (IOException ioe) {
+      LOG.error("Error message: " + ioe.getLocalizedMessage());
+    } finally {
+      if (null != brWriter) {
+        brWriter.flush();
+      }
+      CarbonUtil.closeStreams(brWriter);
+      fileWrite.close();
+    }
+
+  }
+
+  /**
+   * compares passed time stamp with status file delete timestamp and
+   * returns latest timestamp from status file if both are not equal
+   * returns null otherwise
+   *
+   * @param completeBlockName
+   * @param timestamp
+   * @return
+   */
+  public String getTimestampForRefreshCache(String completeBlockName, String timestamp) {
+    long cacheTimestamp = 0;
+    if (null != timestamp) {
+      cacheTimestamp = CarbonUpdateUtil.getTimeStampAsLong(timestamp);
+    }
+    String blockName = CarbonTablePath.addDataPartPrefix(CarbonUpdateUtil.getBlockName(
+        CarbonUpdateUtil.getRequiredFieldFromTID(completeBlockName, TupleIdEnum.BLOCK_ID)));
+    String segmentId =
+        CarbonUpdateUtil.getRequiredFieldFromTID(completeBlockName, TupleIdEnum.SEGMENT_ID);
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+    SegmentUpdateDetails[] listOfSegmentUpdateDetailsArray =
+        readLoadMetadata();
+    for (SegmentUpdateDetails block : listOfSegmentUpdateDetailsArray) {
+      if (segmentId.equalsIgnoreCase(block.getSegmentName()) && block.getBlockName()
+          .equalsIgnoreCase(blockName) && !CarbonUpdateUtil.isBlockInvalid(block.getStatus())) {
+        long deleteTimestampFromStatusFile = block.getDeleteDeltaEndTimeAsLong();
+        if (Long.compare(deleteTimestampFromStatusFile, cacheTimestamp) == 0) {
+          return null;
+        } else {
+          return block.getDeleteDeltaEndTimestamp();
+        }
+      }
+    }
+    return null;
+  }
+
+  /**
+   * This method closes the streams
+   *
+   * @param streams - streams to close.
+   */
+  private void closeStreams(Closeable... streams) {
+    // Added if to avoid NullPointerException in case one stream is being passed as null
+    if (null != streams) {
+      for (Closeable stream : streams) {
+        if (null != stream) {
+          try {
+            stream.close();
+          } catch (IOException e) {
+            LOG.error("Error while closing stream" + stream);
+          }
+        }
+      }
+    }
+  }
+  /**
+   * Get the invalid tasks in that segment.
+   * @param segmentId
+   * @return
+   */
+  public List<String> getInvalidBlockList(String segmentId) {
+
+    // get the original fact file timestamp from the table status file.
+    List<String> listOfInvalidBlocks = new ArrayList<String>();
+    SegmentStatusManager ssm = new SegmentStatusManager(absoluteTableIdentifier);
+    CarbonTablePath carbonTablePath = CarbonStorePath
+        .getCarbonTablePath(absoluteTableIdentifier.getStorePath(),
+            absoluteTableIdentifier.getCarbonTableIdentifier());
+    LoadMetadataDetails[] segmentDetails =
+        ssm.readLoadMetadata(carbonTablePath.getMetadataDirectoryPath());
+    long timestampOfOriginalFacts = 0;
+
+    String startTimestampOfUpdate = "" ;
+    String endTimestampOfUpdate = "";
+
+    for(LoadMetadataDetails segment : segmentDetails){
+      // find matching segment and return timestamp.
+      if (segment.getLoadName().equalsIgnoreCase(segmentId)) {
+        timestampOfOriginalFacts = segment.getLoadStartTime();
+        startTimestampOfUpdate = segment.getUpdateDeltaStartTimestamp();
+        endTimestampOfUpdate = segment.getUpdateDeltaEndTimestamp();
+      }
+    }
+
+    if (startTimestampOfUpdate.isEmpty()) {
+      return listOfInvalidBlocks;
+
+    }
+
+    // now after getting the original fact timestamp, what ever is remaining
+    // files need to cross check it with table status file.
+
+    // filter out the fact files.
+
+    String segmentPath = carbonTablePath.getCarbonDataDirectoryPath("0", segmentId);
+    CarbonFile segDir =
+        FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+
+    final Long endTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(endTimestampOfUpdate);
+    final Long startTimeStampFinal = CarbonUpdateUtil.getTimeStampAsLong(startTimestampOfUpdate);
+    final Long timeStampOriginalFactFinal =
+        timestampOfOriginalFacts;
+
+    CarbonFile[] files = segDir.listFiles(new CarbonFileFilter() {
+
+      @Override public boolean accept(CarbonFile pathName) {
+        String fileName = pathName.getName();
+        if (fileName.endsWith(CarbonCommonConstants.UPDATE_DELTA_FILE_EXT)) {
+          String firstPart = fileName.substring(0, fileName.indexOf('.'));
+
+          long timestamp = Long.parseLong(firstPart
+              .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
+                  firstPart.length()));
+          if (Long.compare(timestamp, endTimeStampFinal) <= 0
+              && Long.compare(timestamp, startTimeStampFinal) >= 0) {
+            return false;
+          }
+          if (Long.compare(timestamp, timeStampOriginalFactFinal) == 0) {
+            return false;
+          }
+          // take the rest of files as they are invalid.
+          return true;
+        }
+        return false;
+      }
+    });
+
+    // gather the task numbers.
+    for(CarbonFile updateFiles : files) {
+      listOfInvalidBlocks.add(updateFiles.getName());
+    }
+
+    return listOfInvalidBlocks;
+  }
+  /**
+   * Returns the invalid timestamp range of a segment.
+   * @param segmentId
+   * @return
+   */
+  public UpdateVO getInvalidTimestampRange(String segmentId) {
+    UpdateVO range = new UpdateVO();
+    for (LoadMetadataDetails segment : segmentDetails) {
+      if (segment.getLoadName().equalsIgnoreCase(segmentId)) {
+        range.setSegmentId(segmentId);
+        range.setFactTimestamp(segment.getLoadStartTime());
+        if (!segment.getUpdateDeltaStartTimestamp().isEmpty() && !segment
+            .getUpdateDeltaEndTimestamp().isEmpty()) {
+          range.setUpdateDeltaStartTimestamp(
+              CarbonUpdateUtil.getTimeStampAsLong(segment.getUpdateDeltaStartTimestamp()));
+          range.setLatestUpdateTimestamp(
+              CarbonUpdateUtil.getTimeStampAsLong(segment.getUpdateDeltaEndTimestamp()));
+        }
+      }
+    }
+    return range;
+  }
+  /**
+   *
+   * @param segmentId
+   * @param block
+   * @param needCompleteList
+   * @return
+   */
+  public CarbonFile[] getDeleteDeltaInvalidFilesList(final String segmentId,
+      final SegmentUpdateDetails block, final boolean needCompleteList,
+      CarbonFile[] allSegmentFiles) {
+
+    final long deltaStartTimestamp =
+        getStartTimeOfDeltaFile(CarbonCommonConstants.DELETE_DELTA_FILE_EXT, block);
+
+    List<CarbonFile> files =
+        new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    for (CarbonFile eachFile : allSegmentFiles) {
+      String fileName = eachFile.getName();
+      if (fileName.endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT)) {
+        String blkName = CarbonTablePath.DataFileUtil.getBlockNameFromDeleteDeltaFile(fileName);
+
+        // complete list of delta files of that block is returned.
+        if (needCompleteList && block.getBlockName().equalsIgnoreCase(blkName)) {
+          files.add(eachFile);
+        }
+
+        // invalid delete delta files only will be returned.
+        long timestamp = CarbonUpdateUtil.getTimeStampAsLong(
+            CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(fileName));
+
+        if (block.getBlockName().equalsIgnoreCase(blkName) && (
+            Long.compare(timestamp, deltaStartTimestamp) < 0)) {
+          files.add(eachFile);
+        }
+      }
+    }
+
+    return files.toArray(new CarbonFile[files.size()]);
+  }
+
+  /**
+   *
+   * @param blockName
+   * @param allSegmentFiles
+   * @return
+   */
+  public CarbonFile[] getAllBlockRelatedFiles(String blockName, CarbonFile[] allSegmentFiles,
+                                              String actualBlockName) {
+    List<CarbonFile> files = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+    for (CarbonFile eachFile : allSegmentFiles) {
+
+      // for carbon data.
+      if (eachFile.getName().equalsIgnoreCase(actualBlockName)) {
+        files.add(eachFile);
+      }
+
+      // get carbon index files of the block.
+      String taskNum = CarbonTablePath.DataFileUtil.getTaskNo(actualBlockName);
+      // String indexFileEndsWith = timestamp + CarbonTablePath.getCarbonIndexExtension();
+      if (eachFile.getName().endsWith(CarbonTablePath.getCarbonIndexExtension()) && eachFile
+          .getName().startsWith(taskNum)) {
+        files.add(eachFile);
+      }
+
+    }
+
+    return files.toArray(new CarbonFile[files.size()]);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/unsafe/CarbonUnsafe.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/unsafe/CarbonUnsafe.java b/core/src/main/java/org/apache/carbondata/core/unsafe/CarbonUnsafe.java
deleted file mode 100644
index 49a9dc4..0000000
--- a/core/src/main/java/org/apache/carbondata/core/unsafe/CarbonUnsafe.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.unsafe;
-
-import java.lang.reflect.Field;
-
-import sun.misc.Unsafe;
-
-
-public final class CarbonUnsafe {
-
-  public static final int BYTE_ARRAY_OFFSET;
-
-  public static final int SHORT_ARRAY_OFFSET;
-
-  public static final int INT_ARRAY_OFFSET;
-
-  public static final int LONG_ARRAY_OFFSET;
-
-  public static final int DOUBLE_ARRAY_OFFSET;
-
-  public static Unsafe unsafe;
-
-  static {
-    try {
-      Field cause = Unsafe.class.getDeclaredField("theUnsafe");
-      cause.setAccessible(true);
-      unsafe = (Unsafe) cause.get(null);
-    } catch (Throwable var2) {
-      unsafe = null;
-    }
-    if (unsafe != null) {
-      BYTE_ARRAY_OFFSET = unsafe.arrayBaseOffset(byte[].class);
-      SHORT_ARRAY_OFFSET = unsafe.arrayBaseOffset(short[].class);
-      INT_ARRAY_OFFSET = unsafe.arrayBaseOffset(int[].class);
-      LONG_ARRAY_OFFSET = unsafe.arrayBaseOffset(long[].class);
-      DOUBLE_ARRAY_OFFSET = unsafe.arrayBaseOffset(double[].class);
-    } else {
-      BYTE_ARRAY_OFFSET = 0;
-      SHORT_ARRAY_OFFSET = 0;
-      INT_ARRAY_OFFSET = 0;
-      LONG_ARRAY_OFFSET = 0;
-      DOUBLE_ARRAY_OFFSET = 0;
-    }
-  }
-}


Mime
View raw message