[GitHub] [carbondata] vikramahuja1001 opened a new pull request #3917: [WIP] clean files refactor

[GitBox]

CarbonDataQA1 commented on pull request #3917:
URL: https://github.com/apache/carbondata/pull/3917#issuecomment-721962218


   Build Success with Spark 2.4.5, Please check CI http://121.244.95.60:12545/job/ApacheCarbon_PR_Builder_2.4.5/3018/
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

QiangCai commented on a change in pull request #3917:
URL: https://github.com/apache/carbondata/pull/3917#discussion_r517813655



##########
File path: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1414,6 +1414,18 @@ private CarbonCommonConstants() {
 
   public static final String BITSET_PIPE_LINE_DEFAULT = "true";
 
+  /**
+   * this is the user defined time(in days), when a specific timestamp subdirectory in
+   * trash folder will expire
+   */
+  @CarbonProperty
+  public static final String CARBON_TRASH_EXPIRATION_DAYS = "carbon.trash.expiration.days";
+
+  /**
+   * Default expiration time of trash folder is 3 days.

Review comment:
       change the comment

##########
File path: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -1414,6 +1414,18 @@ private CarbonCommonConstants() {
 
   public static final String BITSET_PIPE_LINE_DEFAULT = "true";
 
+  /**
+   * this is the user defined time(in days), when a specific timestamp subdirectory in
+   * trash folder will expire
+   */
+  @CarbonProperty
+  public static final String CARBON_TRASH_EXPIRATION_DAYS = "carbon.trash.expiration.days";

Review comment:
       how about "carbon.trash.retention.days"

##########
File path: core/src/main/java/org/apache/carbondata/core/util/CleanFilesUtil.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+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.index.IndexStoreManager;
+import org.apache.carbondata.core.index.Segment;
+import org.apache.carbondata.core.index.TableIndex;
+import org.apache.carbondata.core.indexstore.PartitionSpec;
+import org.apache.carbondata.core.metadata.SegmentFileStore;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the code used in clean files command, to delete the load folders and move the data
+ * to trash folder
+ */
+public final class CleanFilesUtil {
+
+  private static final Logger LOGGER =
+      LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  public static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] newAddedLoadHistoryList, boolean isForceDelete,
+      List<PartitionSpec> specs) {
+    LoadMetadataDetails[] currentDetails =
+        SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath());
+    physicalDataAndIndexDeletion(carbonTable,
+        currentDetails,
+        isForceDelete,
+        specs,
+        currentDetails);
+    if (newAddedLoadHistoryList != null && newAddedLoadHistoryList.length > 0) {
+      physicalDataAndIndexDeletion(carbonTable,
+          newAddedLoadHistoryList,
+          isForceDelete,
+          specs,
+          currentDetails);
+    }
+  }
+
+  /**
+   * Delete the invalid data physically from table.
+   * @param carbonTable table
+   * @param loadDetails Load details which need clean up
+   * @param isForceDelete is Force delete requested by user
+   * @param specs Partition specs
+   * @param currLoadDetails Current table status load details which are required for update manager.
+   */
+  private static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] loadDetails, boolean isForceDelete, List<PartitionSpec> specs,
+      LoadMetadataDetails[] currLoadDetails) {
+    List<TableIndex> indexes = new ArrayList<>();
+    try {
+      for (TableIndex index : IndexStoreManager.getInstance().getAllCGAndFGIndexes(carbonTable)) {
+        if (index.getIndexSchema().isIndex()) {
+          indexes.add(index);
+        }
+      }
+    } catch (IOException e) {
+      LOGGER.warn(String.format(
+          "Failed to get indexes for %s.%s, therefore the index files could not be cleaned.",
+          carbonTable.getAbsoluteTableIdentifier().getDatabaseName(),
+          carbonTable.getAbsoluteTableIdentifier().getTableName()));

Review comment:
       change to
   `carbonTable.getDatabaseName(), carbonTable.getTableName()));`

##########
File path: core/src/main/java/org/apache/carbondata/core/util/CleanFilesUtil.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+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.index.IndexStoreManager;
+import org.apache.carbondata.core.index.Segment;
+import org.apache.carbondata.core.index.TableIndex;
+import org.apache.carbondata.core.indexstore.PartitionSpec;
+import org.apache.carbondata.core.metadata.SegmentFileStore;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the code used in clean files command, to delete the load folders and move the data
+ * to trash folder
+ */
+public final class CleanFilesUtil {
+
+  private static final Logger LOGGER =
+      LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  public static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] newAddedLoadHistoryList, boolean isForceDelete,
+      List<PartitionSpec> specs) {
+    LoadMetadataDetails[] currentDetails =
+        SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath());
+    physicalDataAndIndexDeletion(carbonTable,
+        currentDetails,
+        isForceDelete,
+        specs,
+        currentDetails);
+    if (newAddedLoadHistoryList != null && newAddedLoadHistoryList.length > 0) {
+      physicalDataAndIndexDeletion(carbonTable,
+          newAddedLoadHistoryList,
+          isForceDelete,
+          specs,
+          currentDetails);
+    }
+  }
+
+  /**
+   * Delete the invalid data physically from table.
+   * @param carbonTable table
+   * @param loadDetails Load details which need clean up
+   * @param isForceDelete is Force delete requested by user
+   * @param specs Partition specs
+   * @param currLoadDetails Current table status load details which are required for update manager.
+   */
+  private static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] loadDetails, boolean isForceDelete, List<PartitionSpec> specs,
+      LoadMetadataDetails[] currLoadDetails) {
+    List<TableIndex> indexes = new ArrayList<>();
+    try {
+      for (TableIndex index : IndexStoreManager.getInstance().getAllCGAndFGIndexes(carbonTable)) {
+        if (index.getIndexSchema().isIndex()) {
+          indexes.add(index);
+        }
+      }
+    } catch (IOException e) {
+      LOGGER.warn(String.format(
+          "Failed to get indexes for %s.%s, therefore the index files could not be cleaned.",
+          carbonTable.getAbsoluteTableIdentifier().getDatabaseName(),
+          carbonTable.getAbsoluteTableIdentifier().getTableName()));
+    }
+    SegmentUpdateStatusManager updateStatusManager =
+        new SegmentUpdateStatusManager(carbonTable, currLoadDetails);
+    for (final LoadMetadataDetails oneLoad : loadDetails) {
+      if (checkIfLoadCanBeDeletedPhysically(oneLoad, isForceDelete)) {
+        try {
+          if (oneLoad.getSegmentFile() != null) {
+            SegmentFileStore.deleteSegment(carbonTable.getAbsoluteTableIdentifier().getTablePath(),
+                new Segment(oneLoad.getLoadName(), oneLoad.getSegmentFile()), specs,
+                updateStatusManager);
+          } else {
+            String path = CarbonTablePath.getSegmentPath(carbonTable.getAbsoluteTableIdentifier()
+                .getTablePath(), oneLoad.getLoadName());

Review comment:
                   String path =
                   CarbonTablePath.getSegmentPath(carbonTable.getTablePath(), oneLoad.getLoadName());

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,

Review comment:
       copyDataToTrashBySegment => copySegmentToTrash
   path => segmentPath
   carbonTablePath => tablePath

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);

Review comment:
       carbonFileList => timestampFolderList

##########
File path: core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
##########
@@ -3441,4 +3443,31 @@ public static void agingTempFolderForIndexServer(long agingTime)throws
       });
     }
   }
+
+  /**
+   * The below method tries to get the segment lock for the given segment.
+   */
+  public static boolean tryGettingSegmentLock(LoadMetadataDetails oneLoad,
+      AbsoluteTableIdentifier absoluteTableIdentifier) {
+    ICarbonLock segmentLock = CarbonLockFactory.getCarbonLockObj(absoluteTableIdentifier,
+            CarbonTablePath.addSegmentPrefix(oneLoad.getLoadName()) + LockUsage.LOCK);
+    boolean canGetSegmentLock;
+    try {
+      if (segmentLock.lockWithRetries()) {
+        LOGGER.info("Acquired segment lock on segment: " + oneLoad.getLoadName() + ". It " +
+                "can be deleted as load is not in progress");
+        canGetSegmentLock = true;
+      } else {
+        LOGGER.info("Load in progress for segment" + oneLoad.getLoadName());
+        canGetSegmentLock = false;
+      }
+    } finally {
+      if (segmentLock.unlock()) {
+        LOGGER.info("Segment lock on segment:" + oneLoad.getLoadName() + " is released");
+      } else {
+        LOGGER.error("Unable to release segment lock on : " + oneLoad.getLoadName());
+      }
+    }
+    return canGetSegmentLock;
+  }

Review comment:
       ```
     /**
      * The below method check whether can lock the specified segment.
      */
     public static boolean canLockSegment(AbsoluteTableIdentifier tableIdentifier, String loadName) {
       ICarbonLock segmentLock = CarbonLockFactory.getCarbonLockObj(tableIdentifier,
           CarbonTablePath.addSegmentPrefix(loadName) + LockUsage.LOCK);
       try {
         if (segmentLock.lockWithRetries()) {
           LOGGER.info(String.format("Acquired lock for segment %s of %s.%s", loadName,
               tableIdentifier.getDatabaseName(), tableIdentifier.getTableName()));
           return true;
         } else {
           LOGGER.info(String.format("Unable to lock segment %s of %s.%s", loadName,
               tableIdentifier.getDatabaseName(), tableIdentifier.getTableName()));
           return false;
         }
       } finally {
         if (segmentLock.unlock()) {
           LOGGER.info(String.format("Segment unlocked successfully for %s of %s.%s", loadName,
               tableIdentifier.getDatabaseName(), tableIdentifier.getTableName()));
         } else {
           LOGGER.error(String.format("Unable to unlock segment %s of %s.%s", loadName,
               tableIdentifier.getDatabaseName(), tableIdentifier.getTableName()));
         }
       }
     }
   ```

##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
##########
@@ -731,4 +731,7 @@ public static void touchDirectory(CarbonFile directory, FsPermission permission)
     return getCarbonFile(path, getConfiguration()).listDirs();
   }
 
+  public static List<CarbonFile> getFolderList(CarbonFile carbonFile) throws IOException {

Review comment:
       no need this method

##########
File path: core/src/main/java/org/apache/carbondata/core/util/CleanFilesUtil.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+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.index.IndexStoreManager;
+import org.apache.carbondata.core.index.Segment;
+import org.apache.carbondata.core.index.TableIndex;
+import org.apache.carbondata.core.indexstore.PartitionSpec;
+import org.apache.carbondata.core.metadata.SegmentFileStore;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the code used in clean files command, to delete the load folders and move the data
+ * to trash folder
+ */
+public final class CleanFilesUtil {
+
+  private static final Logger LOGGER =
+      LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  public static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] newAddedLoadHistoryList, boolean isForceDelete,
+      List<PartitionSpec> specs) {
+    LoadMetadataDetails[] currentDetails =
+        SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath());
+    physicalDataAndIndexDeletion(carbonTable,
+        currentDetails,
+        isForceDelete,
+        specs,
+        currentDetails);
+    if (newAddedLoadHistoryList != null && newAddedLoadHistoryList.length > 0) {
+      physicalDataAndIndexDeletion(carbonTable,
+          newAddedLoadHistoryList,
+          isForceDelete,
+          specs,
+          currentDetails);
+    }
+  }
+
+  /**
+   * Delete the invalid data physically from table.
+   * @param carbonTable table
+   * @param loadDetails Load details which need clean up
+   * @param isForceDelete is Force delete requested by user
+   * @param specs Partition specs
+   * @param currLoadDetails Current table status load details which are required for update manager.
+   */
+  private static void physicalDataAndIndexDeletion(CarbonTable carbonTable,
+      LoadMetadataDetails[] loadDetails, boolean isForceDelete, List<PartitionSpec> specs,
+      LoadMetadataDetails[] currLoadDetails) {
+    List<TableIndex> indexes = new ArrayList<>();
+    try {
+      for (TableIndex index : IndexStoreManager.getInstance().getAllCGAndFGIndexes(carbonTable)) {
+        if (index.getIndexSchema().isIndex()) {
+          indexes.add(index);
+        }
+      }
+    } catch (IOException e) {
+      LOGGER.warn(String.format(
+          "Failed to get indexes for %s.%s, therefore the index files could not be cleaned.",
+          carbonTable.getAbsoluteTableIdentifier().getDatabaseName(),
+          carbonTable.getAbsoluteTableIdentifier().getTableName()));
+    }
+    SegmentUpdateStatusManager updateStatusManager =
+        new SegmentUpdateStatusManager(carbonTable, currLoadDetails);
+    for (final LoadMetadataDetails oneLoad : loadDetails) {
+      if (checkIfLoadCanBeDeletedPhysically(oneLoad, isForceDelete)) {
+        try {
+          if (oneLoad.getSegmentFile() != null) {
+            SegmentFileStore.deleteSegment(carbonTable.getAbsoluteTableIdentifier().getTablePath(),

Review comment:
       carbonTable.getTablePath()

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
##########
@@ -41,12 +41,14 @@
   private static final String DICTIONARY_EXT = ".dict";
   public static final String SCHEMA_FILE = "schema";
   private static final String FACT_DIR = "Fact";
+  public static final String CARBON_TRASH_FOLDER_NAME = ".Trash";

Review comment:
       CARBON_TRASH_FOLDER_NAME => TRASH_DIR

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }

Review comment:
       CarbonUtil.closeStreams(dataInputStream, dataOutputStream);

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
##########
@@ -41,12 +41,14 @@
   private static final String DICTIONARY_EXT = ".dict";
   public static final String SCHEMA_FILE = "schema";
   private static final String FACT_DIR = "Fact";
+  public static final String CARBON_TRASH_FOLDER_NAME = ".Trash";
   public static final String SEGMENT_PREFIX = "Segment_";
   private static final String PARTITION_PREFIX = "Part";
   private static final String DATA_PART_PREFIX = "part-";
   public static final String BATCH_PREFIX = "_batchno";
   private static final String LOCK_DIR = "LockFiles";
 
+  public static final String SEGMENTS_METADATA_FOLDER = "segments";

Review comment:
       SEGMENTS_METADATA_FOLDER => SEGMENTS_METADATA_DIR

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,

Review comment:
       copyDataToTrashFolderByFile => copyFileToTrashFolder
   carbonTablePath =>  tablePath
   pathOfFileToCopy => filePathToCopy

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);

Review comment:
       **throw the exception to stop clean this file, if not, it will delete this file by mistake**

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +

Review comment:
       trashFolderPath => trashFilePathSuffix

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);

Review comment:
       **throw the exception to stop clean this segment, if not, it will delete this segment by mistake**
   
   

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {

Review comment:
       carbonFile = > timestampFolder

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {
+          long givenTime = Long.parseLong(carbonFile.getName());
+          // If the timeStamp at which the timeStamp subdirectory has expired as per the user
+          // defined value, delete the complete timeStamp subdirectory
+          if (givenTime + timeStamp < currentTime) {

Review comment:
       if givenTime + timeStamp happened long overflow, it will empty trash
   
   better to calculate an old time outside of for loop, here only compare them.

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {
+          long givenTime = Long.parseLong(carbonFile.getName());
+          // If the timeStamp at which the timeStamp subdirectory has expired as per the user
+          // defined value, delete the complete timeStamp subdirectory
+          if (givenTime + timeStamp < currentTime) {
+            deleteDataFromTrashFolder(carbonFile);
+          } else {
+            LOGGER.info("Timestamp folder not expired yet: " + carbonFile.getAbsolutePath());
+          }
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes all the files and folders in the trash folder of a carbon table.
+   */
+  public static void deleteAllDataFromTrashFolder(String carbonTablePath)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // if the trash folder exists delete the contents of the trash folder
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        for (CarbonFile carbonFile : carbonFileList) {
+          deleteDataFromTrashFolder(carbonFile);
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }

Review comment:
       invoke FileFactory.deleteAllCarbonFilesOfDir(trashPath);

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);

Review comment:
       pathOfTrashFolder => trashPath

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {
+          long givenTime = Long.parseLong(carbonFile.getName());
+          // If the timeStamp at which the timeStamp subdirectory has expired as per the user
+          // defined value, delete the complete timeStamp subdirectory
+          if (givenTime + timeStamp < currentTime) {
+            deleteDataFromTrashFolder(carbonFile);
+          } else {
+            LOGGER.info("Timestamp folder not expired yet: " + carbonFile.getAbsolutePath());
+          }
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes all the files and folders in the trash folder of a carbon table.
+   */
+  public static void deleteAllDataFromTrashFolder(String carbonTablePath)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // if the trash folder exists delete the contents of the trash folder
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        for (CarbonFile carbonFile : carbonFileList) {
+          deleteDataFromTrashFolder(carbonFile);
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes carbonFiles from the trash folder, if it is a directory, it
+   * will delete recursively
+   */
+  private static void deleteDataFromTrashFolder(CarbonFile carbonFile) {

Review comment:
       use the FileFactory.deleteAllCarbonFilesOfDir method
   remove this method

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)

Review comment:
       deleteAllDataFromTrashFolderByTimeStamp => cleanExpiredDataFromTrash
   carbonTablePath => tablePath
   timeStamp => retentionMilliSeconds

##########
File path: docs/configuration-parameters.md
##########
@@ -49,6 +49,8 @@ This section provides the details of all the configurations required for the Car
 | carbon.fs.custom.file.provider | None | To support FileTypeInterface for configuring custom CarbonFile implementation to work with custom FileSystem. |
 | carbon.timeseries.first.day.of.week | SUNDAY | This parameter configures which day of the week to be considered as first day of the week. Because first day of the week will be different in different parts of the world. |
 | carbon.enable.tablestatus.backup | false | In cloud object store scenario, overwriting table status file is not an atomic operation since it uses rename API. Thus, it is possible that table status is corrupted if process crashed when overwriting the table status file. To protect from file corruption, user can enable this property. |
+| carbon.trash.expiration.time | 3 | This parameter specifies the number of days after which the timestamp based subdirectories are expired in the trash folder.|

Review comment:
       change this configuration name and the default value

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {
+          long givenTime = Long.parseLong(carbonFile.getName());
+          // If the timeStamp at which the timeStamp subdirectory has expired as per the user
+          // defined value, delete the complete timeStamp subdirectory
+          if (givenTime + timeStamp < currentTime) {
+            deleteDataFromTrashFolder(carbonFile);
+          } else {
+            LOGGER.info("Timestamp folder not expired yet: " + carbonFile.getAbsolutePath());
+          }
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes all the files and folders in the trash folder of a carbon table.
+   */
+  public static void deleteAllDataFromTrashFolder(String carbonTablePath)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // if the trash folder exists delete the contents of the trash folder
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        for (CarbonFile carbonFile : carbonFileList) {
+          deleteDataFromTrashFolder(carbonFile);
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes carbonFiles from the trash folder, if it is a directory, it
+   * will delete recursively
+   */
+  private static void deleteDataFromTrashFolder(CarbonFile carbonFile) {
+    try {
+      FileFactory.deleteAllCarbonFilesOfDir(carbonFile);
+      LOGGER.info("delete file from trash+ " + carbonFile.getPath());
+    } catch (CarbonFileException e) {
+      LOGGER.error("Error during deleting from trash folder", e);
+    }
+  }
+
+  /**
+   * The below method will list all files in the trash folder
+   */
+  public static List<String> listSegmentsInTrashFolder(String carbonTablePath)

Review comment:
       why need this method?

##########
File path: core/src/main/java/org/apache/carbondata/core/util/path/TrashUtil.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.util.path;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+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.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+import org.apache.commons.io.FileUtils;
+
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.log4j.Logger;
+
+/**
+ * Mantains the trash folder in carbondata. This class has methods to copy data to the trash and
+ * remove data from the trash.
+ */
+public final class TrashUtil {
+
+  private static final Logger LOGGER =
+          LogServiceFactory.getLogService(CarbonUtil.class.getName());
+
+  /**
+   * The below method copies the complete a file to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param pathOfFileToCopy the files which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashFolderByFile(String carbonTablePath, String pathOfFileToCopy,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    CarbonFile carbonFile  = FileFactory.getCarbonFile(pathOfFileToCopy);
+    DataOutputStream dataOutputStream = null;
+    DataInputStream dataInputStream = null;
+    try {
+      if (carbonFile.exists()) {
+        // FileUtils.copyFileToDirectory(new File(pathOfFileToCopy), new File(trashFolderPath));
+        if (!new File(trashFolderPath).exists()) {
+          FileFactory.createDirectoryAndSetPermission(trashFolderPath,
+              new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+        }
+        dataOutputStream = FileFactory
+          .getDataOutputStream(trashFolderPath + CarbonCommonConstants.FILE_SEPARATOR +
+            carbonFile.getName());
+        dataInputStream = FileFactory
+          .getDataInputStream(pathOfFileToCopy);
+        IOUtils.copyBytes(dataInputStream, dataOutputStream, CarbonCommonConstants.BYTEBUFFER_SIZE);
+        LOGGER.info("File: " + pathOfFileToCopy + " successfully copied to the trash folder: "
+            + trashFolderPath);
+      }
+    } catch (IOException exception) {
+      LOGGER.error("Unable to copy " + pathOfFileToCopy + " to the trash folder", exception);
+    } finally {
+      try {
+        CarbonUtil.closeStream(dataInputStream);
+        CarbonUtil.closeStream(dataOutputStream);
+      } catch (IOException exception) {
+        LOGGER.error(exception.getMessage(), exception);
+      }
+    }
+  }
+
+  /**
+   * The below method copies the complete segment folder to the trash folder. Provide necessary
+   * timestamp and the segment number in the suffixToAdd  variable, so that the proper folder is
+   * created in the trash folder.
+   *
+   * @param carbonTablePath table path of the carbon table
+   * @param path the folder which are to be moved to the trash folder
+   * @param suffixToAdd timestamp, partition folder(if any) and segment number
+   * @return
+   */
+  public static void copyDataToTrashBySegment(CarbonFile path, String carbonTablePath,
+      String suffixToAdd) {
+    String trashFolderPath = CarbonTablePath.getTrashFolderPath(carbonTablePath) +
+        CarbonCommonConstants.FILE_SEPARATOR + suffixToAdd;
+    try {
+      FileUtils.copyDirectory(new File(path.getAbsolutePath()), new File(trashFolderPath));
+      LOGGER.info("Segment: " + path.getAbsolutePath() + " has been copied to the trash folder" +
+          " successfully");
+    } catch (IOException e) {
+      LOGGER.error("Unable to create the trash folder and copy data to it", e);
+    }
+  }
+
+  /**
+   * The below method deletes timestamp subdirectories in the trash folder which have expired as
+   * per the user defined expiration time
+   */
+  public static void deleteAllDataFromTrashFolderByTimeStamp(String carbonTablePath, long timeStamp)
+          throws IOException {
+    String pathOfTrashFolder = CarbonTablePath.getTrashFolderPath(carbonTablePath);
+    // Deleting the timestamp based subdirectories in the trashfolder by the given timestamp.
+    if (FileFactory.isFileExist(pathOfTrashFolder)) {
+      try {
+        List<CarbonFile> carbonFileList = FileFactory.getFolderList(pathOfTrashFolder);
+        long currentTime = new Timestamp(System.currentTimeMillis()).getTime();
+        for (CarbonFile carbonFile : carbonFileList) {
+          long givenTime = Long.parseLong(carbonFile.getName());
+          // If the timeStamp at which the timeStamp subdirectory has expired as per the user
+          // defined value, delete the complete timeStamp subdirectory
+          if (givenTime + timeStamp < currentTime) {
+            deleteDataFromTrashFolder(carbonFile);
+          } else {
+            LOGGER.info("Timestamp folder not expired yet: " + carbonFile.getAbsolutePath());
+          }
+        }
+      } catch (IOException e) {
+        LOGGER.error("Error during deleting from trash folder", e);
+      }
+    }
+  }
+
+  /**
+   * The below method deletes all the files and folders in the trash folder of a carbon table.
+   */
+  public static void deleteAllDataFromTrashFolder(String carbonTablePath)

Review comment:
       deleteAllDataFromTrashFolder => emptyTrash
   carbonTablePath => tablePath




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

vikramahuja1001 commented on pull request #3917:
URL: https://github.com/apache/carbondata/pull/3917#issuecomment-722256333


   > please split this PR to two PRs
   >
   > 1. one PR for data trash
   > 2. another PR for dry run
   >
   > and we need third PR to refactor clean files
   
   OKay, so this PR just for trash , i will change the scope, description and code


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

vikramahuja1001 closed pull request #3917:
URL: https://github.com/apache/carbondata/pull/3917


   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]